系列五（实战）、发送 & 接收同步消息（Java操作RocketMQ）

一、发送 & 接收同步消息

1.1、前置说明

????????Java操作RocketMQ系列文章是基于Maven项目进行演示的，所以大家只需要新建一个普通的Maven项目，然后将pom、常量信息粘进自己的代码中即可！后续会补充SpringBoot中如何操作RocketMQ，万事万物都是先易后难，先简单后复杂。不说那么多了，下面通过实战案例为大家演示，如果帮助到了你，欢迎点赞 ?+ 评论 +?关注 + 转发 + 收藏，让我们一起进步！

1.2、项目概览

1.3、公共代码

1.3.1、pom

<dependencies>
	<dependency>
	  <groupId>org.apache.rocketmq</groupId>
	  <artifactId>rocketmq-client</artifactId>
	  <version>4.5.0</version>
	</dependency>

	<dependency>
	  <groupId>junit</groupId>
	  <artifactId>junit</artifactId>
	  <version>4.13.2</version>
	  <scope>test</scope>
	</dependency>
	<dependency>
	  <groupId>com.alibaba</groupId>
	  <artifactId>fastjson</artifactId>
	  <version>2.0.25</version>
	</dependency>
	<dependency>
	  <groupId>cn.hutool</groupId>
	  <artifactId>hutool-all</artifactId>
	  <version>5.8.16</version>
	</dependency>
	<dependency>
	  <groupId>org.apache.commons</groupId>
	  <artifactId>commons-collections4</artifactId>
	  <version>4.4</version>
	</dependency>
	<dependency>
	  <groupId>org.apache.commons</groupId>
	  <artifactId>commons-lang3</artifactId>
	  <version>3.12.0</version>
	</dependency>

	<!-- 普通maven项目中使用Sl4j注解 -->
	<dependency>
	  <groupId>org.projectlombok</groupId>
	  <artifactId>lombok</artifactId>
	  <version>1.18.22</version>
	</dependency>
	<dependency>
	  <groupId>org.slf4j</groupId>
	  <artifactId>slf4j-api</artifactId>
	  <version>1.7.32</version>
	</dependency>
	<dependency>
	  <groupId>ch.qos.logback</groupId>
	  <artifactId>logback-classic</artifactId>
	  <version>1.2.10</version>
	</dependency>
</dependencies>

1.3.2、RocketMQConstant

/**
 * @Author: 一叶浮萍归大海
 * @Date: 2023/12/25 09:13
 * @Description: RocketMQ常量信息
 */
public class RocketMQConstant {

    /**
     * 配置RocketMQ NameSrv的地址
     */
    public static final String NAME_SERVER_ADDR = "<your MQ IP>:9876";

}

1.4、发送 & 接收同步消息

1.4.1、消息模型

1.4.2、特点

? ? ? ? 同步消息有一个最大的特点是：生产者发送完消息后，服务端需返回一个消息给生产者，用于告知生产者当前消息的状态，这种方式非常的安全，但是性能上却没那么高了，而且在RocketMQ的集群模式下，需要等到所有的从机都复制了消息之后才会返回，所以一般对于重要的消息采用同步消息比较好。

1.4.3、Demo01MQTestApp

/**
 * @Author : 一叶浮萍归大海
 * @Date: 2023/12/24 23:47
 * @Description: 发送 & 接收同步消息
 */
public class Demo01MQTestApp {


    /**
     * 发送同步消息
     */
    @Test
    public void demo1Producer() throws Exception {
        // 1、创建一个生产者
        DefaultMQProducer producer = new DefaultMQProducer("sync-producer-group");

        // 2、连接NameServer
        producer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);

        // 3、启动
        producer.start();

        // 4、创建消息
        Message message = new Message("sync-Topic","这是一个同步消息".getBytes(StandardCharsets.UTF_8));

        // 5、发送消息
        SendResult result = producer.send(message);
        System.out.println("result = " + JSON.toJSONString(result));
        System.out.println("生产者demo1Producer start success!");

        // 6、关闭生产者
        producer.shutdown();
    }
    /**
     * 接收同步消息（Push方式）
     */
    @Test
    public void demo1PushConsumer() throws Exception {
        // 1、创建一个消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sync-consumer-group");
        // 2、连接NameServer
        consumer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);
        // 3、订阅消息，*表示订阅该主题所有的消息
        consumer.subscribe("sync-Topic","*");
        // 4、设置监听器（采用异步回调方式，一直监听）
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 业务逻辑
                System.out.println("我是消费者demo1PushConsumer");
                for (MessageExt msg : msgs) {
                    System.out.println("我是消费者demo1PushConsumer，我收到了消息，消息详情：" + JSON.toJSONString(msg.toString()));
                    System.out.println("我是消费者demo1PushConsumer，我收到了消息，消息内容：" + new String(msg.getBody()));
                }
                System.out.println("消息上下文：" + context);

                /**
                 * 返回值：消费消息成功与否
                 *      CONSUME_SUCCESS：表明消费成功，消息会从MQ出队
                 *      RECONSUME_LATER：表明消费失败，消息会重新回到队里，过一会儿再重新投递出来给当前消费者或者其他消费者
                 */
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 5、启动
        consumer.start();
        System.out.println("demo1PushConsumer start success!");

        // 6、挂起当前JVM
        System.in.read();
    }

}

1.4.4、测试

先后运行demo1PushConsumer和demo1Producer，观察控制台日志：