（一）整体结构

• 父级pom

<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>priv.wei</groupId><artifactId>rocketmq</artifactId><version>0.0.1-SNAPSHOT</version><packaging>pom</packaging><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.6.RELEASE</version><relativePath /></parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><java.version>1.8</java.version><fastjson.version>1.2.62</fastjson.version><commons-lang3.version>3.9</commons-lang3.version><rocketmq.version>4.3.2</rocketmq.version><maven-jar-plugin.version>3.1.1</maven-jar-plugin.version></properties><dependencyManagement><dependencies><!-- 公共jar版本管理 start --><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>${commons-lang3.version}</version></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId><version>${rocketmq.version}</version></dependency><!-- 公共jar版本管理 end --></dependencies></dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency><dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId></dependency><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-client</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><optional>true</optional></dependency></dependencies><!--指定使用maven打包 --><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>${java.version}</source><target>${java.version}</target></configuration></plugin><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-surefire-plugin</artifactId><configuration><skipTests>true</skipTests>    <!--默认关掉单元测试 --></configuration></plugin><plugin><artifactId>maven-compiler-plugin</artifactId><configuration><source>1.8</source><target>1.8</target><encoding>UTF-8</encoding><optimize>true</optimize><debug>false</debug><showDeprecation>true</showDeprecation><showWarnings>false</showWarnings><compilerArguments><verbose /><bootclasspath>${java.home}/lib/rt.jar;${java.home}/lib/jce.jar</bootclasspath></compilerArguments></configuration></plugin></plugins></build><modules><module>producer</module><module>consumer</module></modules>
</project>

（二）生产者模块

• Producer.java

package priv.wei.producer.mq;import javax.annotation.PostConstruct;import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import lombok.extern.slf4j.Slf4j;
import priv.wei.producer.mq.config.MqConfig;
import priv.wei.producer.util.ExpUtil;/*** 生产者*/
@Slf4j
@Component
public class Producer {@Autowiredprivate MqConfig mc;private DefaultMQProducer producer;/*** 系统启动时自动初始化生产者* @throws MQClientException*/@PostConstructprivate void init() throws MQClientException{//示例生产者producer = new DefaultMQProducer(mc.getProducerGroup());//不开启vip通道 开通口端口会减2producer.setVipChannelEnabled(false);//绑定name serverproducer.setNamesrvAddr(mc.getNameServer());//初始化启动producer.start();log.info("mq生产者初始化成功！");}/*** 发送请求* @param topic 主题* @param tag 标签* @param content 消息内容*/public void send(String topic,String tag,String content){try {Message msg = new Message(topic, tag, content.getBytes(RemotingHelper.DEFAULT_CHARSET));SendResult sr = producer.send(msg);log.info("发送响应：MsgId:{},发送状态SendStatus:{}",sr.getMsgId(),sr.getSendStatus());} catch (Exception e) {log.error("发送信息异常:{}",ExpUtil.getStackMsg(e));}}/*** 单独获取生产者* @return DefaultMQProducer*/public DefaultMQProducer get(){return this.producer;}/*** 关闭生产者*/public void shutdown(){this.producer.shutdown();}
}

（三）消费者模块

• TestDateConsumer.java

package priv.wei.consumer.mq.listener;import javax.annotation.PostConstruct;import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
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.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;import lombok.extern.slf4j.Slf4j;
import priv.wei.consumer.mq.config.MqConfig;
import priv.wei.consumer.service.ConsumerService;
import priv.wei.consumer.util.ExpUtil;/*** 监听消息，并消费*/
@Slf4j
@Component
public class TestDateConsumer {@Autowiredprivate MqConfig mc;@Autowiredprivate ConsumerService consumerService;private DefaultMQPushConsumer consumer;/*** 项目启动时初始化监听test_topic-test_tag*/@PostConstructprivate void init(){try {consumer = new DefaultMQPushConsumer();consumer.setNamesrvAddr(mc.getNameServer());consumer.setConsumerGroup(mc.getConsumerGroup());//监听指定的topic，tag消息consumer.subscribe("test_topic", "test_tag");consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {try {for (MessageExt messageExt : list) {//消费内容consumerService.test(new String(messageExt.getBody()));}} catch (Exception e) {log.info("消费异常：{}",ExpUtil.getStackMsg(e));return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功});consumer.start();log.info("初始化监听：topic:test_topic,tag:test_tag启动成功！");} catch (Exception e) {log.error("初始化监听：topic:test_topic,tag:test_tag启动异常：{}",ExpUtil.getStackMsg(e));}}
}

（四）代码GitHub地址

https://github.com/shouwangying/rocketmq-demo.git