Java操作 RabbitMQ 示例

前言

• 此 demo 主要演示了 Spring Boot 如何集成 RabbitMQ，并且演示了基于直接队列模式、分列模式、主题模式的消息发送和接收。

正文

pom文件示例

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <artifactId>spring-boot-demo-mq-rabbitmq</artifactId>
    <version>1.0.0-SNAPSHOT</version>
    <packaging>jar</packaging>

    <name>spring-boot-demo-mq-rabbitmq</name>
    <description>Demo project for Spring Boot</description>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
            <version>1.16.20</version>
        </dependency>

        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>4.6.3</version>
        </dependency>

        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>28.1-jre</version>
        </dependency>
    </dependencies>

    <build>
        <finalName>spring-boot-demo-mq-rabbitmq</finalName>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

application.yml

server:
  port: 8080
  servlet:
    context-path: /demo
spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /
    # 手动提交消息
    listener:
      simple:
        acknowledge-mode: manual
      direct:
        acknowledge-mode: manual

RabbitConsts.java

package cloud.pm.log.common.context;

/**
 * 功能描述：
 * 【RabbitMQ常量池】
 *
 * @author chihiro
 * @version V1.0
 * @date 2019/09/03 19:56
 */
public class RabbitConstants {

    /**
     * 分列模式
     */
    public final static String FANOUT_MODE_QUEUE = "fanout.mode";

    /**
     * 日志打印队列
     */
    public final static String QUEUE_LOG_PRINT = "queue.log.recode";

    /**
     * 主题模式
     */
    public final static String TOPIC_MODE_QUEUE = "topic.mode";

    /**
     * 主题模式
     */
    public final static String TOPIC_ROUTING_KEY = "topic.*";

}

RabbitMqConfig.java

package cloud.pm.log.common.config;

import cloud.pm.log.common.context.RabbitConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 功能描述：
 * 【RabbitMQ配置，主要是配置队列，如果提前存在该队列，可以省略本配置类】
 *
 * @author chihiro
 * @version V1.0
 * @date 2019/09/03 19:58
 */
@Slf4j
@Configuration
public class RabbitMqConfig {

    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> log.info("消息发送成功:correlationData[{}],ack[{}],cause[{}]", correlationData, ack, cause));
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> log.info("消息丢失:exchange[{}],route[{}],replyCode[{}],replyText[{}],message:{}", exchange, routingKey, replyCode, replyText, message));
        return rabbitTemplate;
    }

    /**
     * 日志打印队列
     */
    @Bean
    public Queue logPrintQueue() {
        return new Queue(RabbitConstants.QUEUE_LOG_PRINT);
    }

    /**
     * 分列模式队列
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(RabbitConstants.FANOUT_MODE_QUEUE);
    }

    /**
     * 分列模式绑定队列
     *
     * @param logPrintQueue  绑定队列
     * @param fanoutExchange 分列模式交换器
     */
    @Bean
    public Binding fanoutBinding(Queue logPrintQueue, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(logPrintQueue).to(fanoutExchange);
    }

    /**
     * 主题队列
     */
    @Bean
    public Queue topicQueue() {
        return new Queue(RabbitConstants.TOPIC_ROUTING_KEY);
    }

    /**
     * 主题模式队列
     * <li>路由格式必须以 . 分隔，比如 user.email 或者 user.aaa.email</li>
     * <li>通配符 * ，代表一个占位符，或者说一个单词，比如路由为 user.*，那么 user.email 可以匹配，但是 user.aaa.email 就匹配不了</li>
     * <li>通配符 # ，代表一个或多个占位符，或者说一个或多个单词，比如路由为 user.#，那么 user.email 可以匹配，user.aaa.email 也可以匹配</li>
     */
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(RabbitConstants.TOPIC_MODE_QUEUE);
    }

    /**
     * 主题模式绑定队列2
     *
     * @param topicQueue    主题队列
     * @param topicExchange 主题模式交换器
     */
    @Bean
    public Binding topicBinding(Queue topicQueue, TopicExchange topicExchange) {
        return BindingBuilder.bind(topicQueue).to(topicExchange).with(RabbitConstants.TOPIC_ROUTING_KEY);
    }

}

RabbitMqHandler.java

package cloud.pm.log.handler;

import cloud.pm.log.common.context.RabbitConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 功能描述：
 * 【消息队列处理器】
 *
 * @author chihiro
 * @version V1.0
 * @date 2019/09/03 20:22
 */
@Slf4j
@Component
public class RabbitMqHandler {

    /**
     * 日志打印处理handler
     *
     * @param message 待处理的消息体
     */
    @RabbitListener(queues = RabbitConstants.QUEUE_LOG_PRINT)
    public void queueLogPrintHandler(String message) {
        log.info("接收到操作日志记录消息：[{}]", message);
    }

    /**
     * 主题模式处理handler
     *
     * @param message 待处理的消息体
     */
    @RabbitListener(queues = RabbitConstants.TOPIC_ROUTING_KEY)
    public void queueTopicHandler(String message) {
        log.info("主题模式处理器，接收消息：[{}]", message);
    }

}

RabbitMqTests.java

package cloud.pm.log;

import cloud.pm.log.common.context.RabbitConstants;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * 功能描述：
 * 【消息队列测试】
 *
 * @author chihiro
 * @version V1.0
 * @date 2019/09/03 20:15
 */
@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class RabbitMqTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     * 测试直接模式发送
     */
    @Test
    public void sendDirect() {
        String message = "direct message";
        rabbitTemplate.convertAndSend(RabbitConstants.QUEUE_LOG_PRINT, message);
        log.info("消息发送成功：[{}]", message);
    }

    /**
     * 测试主题模式发送
     */
    @Test
    public void sendTopic() {
        String message = "topic message";
        rabbitTemplate.convertAndSend(RabbitConstants.TOPIC_MODE_QUEUE, "topic.queue", message);
        log.info("消息发送成功：[{}]", message);
    }


}

• 以上示例代码编写完毕

后记

• 这套代码是博主从网上找来调整为自己使用的，非博主独立构建。
• 若遇到其他Bug，请通过博客内联系方式找博主修复。