一、概述

        同步消息的特征是消息发出后会有一个返回值，即RocketMQ服务器收到消息后的一个确认，这种方式非常安全，但是性能上却没有那么高，而且在集群模式下，也是要等到所有的从机都复制了消息以后才会返回，适用于重要的消息传递，例如：短信通知

二、案例代码

2.1、pom.xml

<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><groupId>org.star</groupId><artifactId>rocketmq-example</artifactId><version>1.0-SNAPSHOT</version><packaging>jar</packaging><name>rocketmq-example</name><url>http://maven.apache.org</url><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><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><!-- 锁定Java编译的版本 --><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.8.1</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin></plugins></build>
</project>

2.2、RocketMQConstant

package org.star.constants;/*** @Author: 一叶浮萍归大海* @Date: 2023/7/27 16:42* @Description:*/
public class RocketMQConstant {/*** 配置RocketMQ NameSrv的地址*/public static final String NAME_SERVER_ADDR = "192.168.173.219:9876";}

2.3、SimpleConsumer

package org.star.simple.consumer;import cn.hutool.core.util.StrUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.star.constants.RocketMQConstant;import java.util.List;/*** @Author: 一叶浮萍归大海* @Date: 2023/8/25 10:20* @Description: 简单消息消费者*/
@Slf4j
public class SimpleConsumer {public static void main(String[] args) throws Exception {/*** 消费消息分两种* （1）拉模式：消费者主动去Broker上拉消息* （2）推模式：消费者等待Broker把消息推送过来*/DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("SimpleMessageGroup");consumer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);consumer.subscribe("SimpleTopic", "*");consumer.setMessageListener(new MessageListenerConcurrently() {@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {if (CollectionUtils.isNotEmpty(list)) {String body = StrUtil.utf8Str(list.get(0).getBody());log.info("收到消息 body:{}",body);}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}});consumer.start();System.out.println("consumer started!");}}

2.4、SyncProducer

package org.star.simple.producer;import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.star.constants.RocketMQConstant;import java.nio.charset.StandardCharsets;/*** @Author: 一叶浮萍归大海* @Date: 2023/8/25 10:12* @Description: 同步发送：等待消息返回后再继续执行下面的操作*/
@Slf4j
public class SyncProducer {public static void main(String[] args) throws Exception {DefaultMQProducer producer = new DefaultMQProducer("SyncProducerGroup");producer.setNamesrvAddr(RocketMQConstant.NAME_SERVER_ADDR);producer.start();for (int i = 0; i < 3; i++) {Message message = new Message("SimpleTopic", ("我是第[" + i + "]个简单消息").getBytes(StandardCharsets.UTF_8));SendResult sendResult = producer.send(message);log.info("第[" + i + "]个简单消息发送成功 sendStatus：{}，msgId：{}，topic：{}", sendResult.getSendStatus(),sendResult.getMsgId(),sendResult.getMessageQueue().getTopic());}producer.shutdown();}}

2.5、控制台打印结果

# 生产者端
12:14:32.339 [main] INFO org.star.simple.producer.SyncProducer - 第[0]个简单消息发送成功 sendStatus：SEND_OK，msgId：C0A81FB25D9418B4AAC207C6D9850000，topic：SimpleTopic
12:14:32.343 [main] INFO org.star.simple.producer.SyncProducer - 第[1]个简单消息发送成功 sendStatus：SEND_OK，msgId：C0A81FB25D9418B4AAC207C6D9940001，topic：SimpleTopic
12:14:32.348 [main] INFO org.star.simple.producer.SyncProducer - 第[2]个简单消息发送成功 sendStatus：SEND_OK，msgId：C0A81FB25D9418B4AAC207C6D9970002，topic：SimpleTopic# 消费者端
12:14:32.337 [ConsumeMessageThread_10] INFO org.star.simple.consumer.SimpleConsumer - 收到消息 body:我是第[0]个简单消息
12:14:32.345 [ConsumeMessageThread_11] INFO org.star.simple.consumer.SimpleConsumer - 收到消息 body:我是第[1]个简单消息
12:14:32.349 [ConsumeMessageThread_12] INFO org.star.simple.consumer.SimpleConsumer - 收到消息 body:我是第[2]个简单消息