SpringBoot整合rocketmq客户端集群，自定义生产者和消费者实例

男娘i 2022-03-20 10:44 1585阅读 0赞

### 简介 ###

*   RocketMQ，是阿里巴巴自研消息中间件产品，使用它有两种方式：

1.   自行搭建rocketmq集群，然后创建rocketmq启动包、生产者和消费者，进行消息推送和消费。这种方式在自己拥有可靠服务器的前提下，可以自己搭建属于自己的rocketmq集群，但是如果服务器性能不太好的话，集群很容易宕机，造成消息丢失的情况；
2.   直接购买阿里云的rocketmq产品，然后利用阿里云提供的demo，创建生产者和消费者，进行消息推送和消费。这种方式价格便宜，进入阿里云控制台维护也相对方便，普遍应用于中小型公司；

*  本篇博客主要讲述，使用springboot框架整合自定义rocketmq客户端，自行搭建rocketmq集群，然后创建生产者和消费者，进行消息推送和消费（这里直接部署一个windows的单机版rocketmq进行测试，单机部署和集群部署都可以参考作者的其他博客，都有详细的说明）：

1.  windows部署rocketmq单机版博客：[https://blog.csdn.net/alan\_liuyue/article/details/86648280][https_blog.csdn.net_alan_liuyue_article_details_86648280]；
2.  linux部署rocketmq集群版博客：[https://blog.csdn.net/alan\_liuyue/article/details/86613896][https_blog.csdn.net_alan_liuyue_article_details_86613896]；

### **实践过程** ###

*  首先，引入rocketmq-starter启动项目包（这里暂时不打包成jar，为方便开发而直接在pom引入启动项目），添加pom依赖；

rocketmq-starter启动包下载地址：[https://download.csdn.net/download/alan\_liuyue/10937415][https_download.csdn.net_download_alan_liuyue_10937415]

**      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.apache.rocketmq</groupId>
    	<artifactId>spring-boot-starter-java-rocketmq-producer</artifactId>
     	<version>1.0.0</version>
    
    	<name>RocketmqProducer</name>
    	<description>rocketmq生产者</description>
    
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    		<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    		<java.version>1.8</java.version>
    		<rocketmq.version>4.3.0</rocketmq.version>
    	</properties>
    
    	<parent>
    		<groupId>org.springframework.boot</groupId>
    		<artifactId>spring-boot-starter-parent</artifactId>
    		<version>1.5.9.RELEASE</version>
    
    	</parent>
    
    	<dependencies>
    		
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-starter-web</artifactId>
    		</dependency>
    		
    		
      		<dependency>
    		    <groupId>org.apache.rocketmq</groupId>
    		    <artifactId>spring-boot-starter-sixmonth-rocketmq</artifactId>
    		    <version>1.0.0-SNAPSHOT</version>
    		</dependency>
    
    		
    		<dependency>
    		    <groupId>org.mybatis.spring.boot</groupId>
    		    <artifactId>mybatis-spring-boot-starter</artifactId>
    		    <version>1.3.2</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>mysql</groupId>
    			<artifactId>mysql-connector-java</artifactId>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>druid</artifactId>
    			<version>1.0.29</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-beans</artifactId>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>fastjson</artifactId>
    			<version>1.2.3</version>
    		</dependency>
    		<dependency>
    			<groupId>org.codehaus.jackson</groupId>
    			<artifactId>jackson-mapper-asl</artifactId>
    			<version>1.9.13</version>
    		</dependency>
    
    		
    		<dependency>
    			<groupId>org.springframework.boot</groupId>
    			<artifactId>spring-boot-devtools</artifactId>
    			<optional>true</optional>
    		</dependency>
    
    
    		
    		<dependency>
    			<groupId>com.google.code.gson</groupId>
    			<artifactId>gson</artifactId>
    		</dependency>		
    		
    		<dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
            </dependency>
            <dependency>
                <groupId>org.assertj</groupId>
                <artifactId>assertj-core</artifactId>
            </dependency>
    
    	</dependencies>
    
        <build>
            <plugins>
                 <plugin>
                    <artifactId>maven-compiler-plugin</artifactId>
                    
                    <configuration>
                        <source>1.8</source>
                        <target>1.8</target>
                        <encoding>UTF-8</encoding>
                        
                    </configuration>
                </plugin>
    
                <plugin>
                    <artifactId>maven-assembly-plugin</artifactId>
                    <configuration>
                        <descriptorRefs>
                            <descriptorRef>jar-with-dependencies</descriptorRef>
                        </descriptorRefs>
                        <archive>
                            <manifest>
                                <mainClass>com.sixmonth.rocketmq.RocketmqApplication</mainClass>
                            </manifest>
                        </archive>
                    </configuration>
                    <executions>
                        <execution>
                            <id>make-assembly</id>
                            <phase>package</phase>
                            <goals>
                                <goal>single</goal>
                            </goals>
                        </execution>
                    </executions>
                </plugin>          
            </plugins>
        </build>
    
    </project>

*  接着，配置rocketmq属性：

#配置rocketmq      
    spring:
      rocketmq:
        nameServer: 127.0.0.1:9876  #mq的nameserver地址,集群使用分号连接
        producer:
          group: sixmonth  #mq生产者组名称，单个应用一般只选取一个
          send-msg-timeout: 5000  #发送超时配置毫秒数, 可选, 默认3000
          retry-times-when-send-failed: 2 #发送消息失败之后重试次数
          max-message-size: 4194304  #消息最大长度,默认1024*1024*4(4M)

*  之后，创建rocketmq生产者生产实例：

package com.rocketmq.service.rocketmq.producer;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import javax.annotation.Resource;
    
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.spring.starter.core.RocketMQTemplate;
    import org.springframework.stereotype.Service;
    
    /**
     * 消息生产者，可与消费者分离
     * @author hqc
     * @Date 2018年12月28日
     *
     */
    @Service
    public class RocketmqProducerService {
    	
       public static final String topic = "temTopic"; //定义主题，可分离放在配置文件里面，一般一个主题发送同类消息，主题格式：`topicName:tags`
    	
       // 声明并引用RocketMQTemplate
       @Resource
       private RocketMQTemplate rocketMQTemplate;
       
       /**
        * 同步发送字符串消息
        * @param msg
        * @return
        */
       public boolean sendMsg(String message) {
    	   // 以同步的方式发送字符串消息给指定的topic
    	   Long startTime = System.currentTimeMillis();
    	   /*int size = 10000;
    	   message = "[";
    	   for(int i=0;i<size;i++) {
    		   String mess = "{id:'123456',rate:'"+msg+i+"'},";
    				   if(i==(size-1)) {
    					   mess = "{id:'123456',rate:'"+msg+i+"'}]";
    				   }
    		   message+=mess;
    	   }*/
    	   SendResult sendResult = rocketMQTemplate.syncSend(topic, message);
           // 打印发送结果信息
           System.out.printf("string-topic syncSend1 sendResult=%s %n", sendResult);
           Long endTime = System.currentTimeMillis();
    	   System.out.println("单次生产耗时："+(endTime-startTime)/1000);
    	   return true;
       }
       
       /**
        * 同步发送实体对象消息
        * 可靠同步发送：同步发送是指消息发送方发出数据后，会在收到接收方发回响应之后才发下一个数据包的通讯方式；
        * 特点：速度快；有结果反馈；数据可靠；
        * 应用场景：应用场景非常广泛，例如重要通知邮件、报名短信通知、营销短信系统等；
        * @param msg
        * @return
        */
       public boolean sendMsgObj() {
    	   Map<String,Object> map = new HashMap<String,Object>();
    	   map.put("name", "sixmonth");
    	   map.put("age", "88");
           Message msg = new Message("openSporting", "tem1", map.toString().getBytes());
    	   // 以同步的方式发送字符串消息给指定的topic
    	   SendResult sendResult = rocketMQTemplate.syncSend(topic, msg);
    	   // 打印发送结果信息
    	   System.out.printf("string-topic syncSend1 sendResult=%s %n", sendResult);
    	   return true;
       }
       
       /**
        * 异步发送消息
        * 可靠异步发送：发送方发出数据后，不等接收方发回响应，接着发送下个数据包的通讯方式；
        * 特点：速度快；有结果反馈；数据可靠；
        * 应用场景：异步发送一般用于链路耗时较长,对 rt响应时间较为敏感的业务场景,例如用户视频上传后通知启动转码服务,转码完成后通知推送转码结果等；
        * @param msg
        * @return
        */
       public boolean sendMsgAsy() {
    	   Map<String,Object> map = new HashMap<String,Object>();
    	   map.put("name", "sixmonth");
    	   map.put("age", "88");
    	   // 以同步的方式发送字符串消息给指定的topic
    	   rocketMQTemplate.asyncSend(topic, map,new SendCallback() {
               @Override
               public void onSuccess(final SendResult sendResult) {
                   //消息发送成功
                   System.out.println("send message success. topic=" + sendResult.getSendStatus());
               }
    			@Override
    			public void onException(Throwable e) {
    				//消息发送失败
    				System.out.println("send message failed. execption=" + e.getStackTrace() );
    			}
    	   });
    	   return true;
       }
       
       /**
        * 单向发送
        * 单向发送：只负责发送消息，不等待服务器回应且没有回调函数触发，即只发送请求不等待应答；此方式发送消息的过程耗时非常短，一般在微秒级别；
        * 特点：速度最快，耗时非常短，毫秒级别；无结果反馈；数据不可靠，可能会丢失；
        * 应用场景：适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集；
        * @return
        */
       public boolean sendMsgOneway() {
    	   Map<String,Object> map = new HashMap<String,Object>();
    	   map.put("name", "sixmonth1");
    	   map.put("age", "88");
    	   rocketMQTemplate.sendOneWay(topic, map);
    	   return true;
       }
       
       
    }

*  最后，创建rocketmq消费者消费实例：

package com.sixmonth.rocketmq.service.rocketmq;
    
    import java.util.ArrayList;
    import java.util.List;
    
    import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
    import org.apache.rocketmq.spring.starter.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.starter.core.RocketMQListener;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    
    import com.alibaba.fastjson.JSONArray;
    import com.alibaba.fastjson.JSONException;
    import com.alibaba.fastjson.JSONObject;
    import com.sixmonth.rocketmq.common.util.Tool;
    
    /**
     * 消息消费者，可与生产者分离
     * 这里定义的messageMode是集群消费，如果是同一消费组则会组内消费者均衡消费；
     * 如果是不同消费组，则会起到广播消费的效果
     * @author hqc
     * @Date 2018年12月28日
     *
     */
    @Service
    @RocketMQMessageListener(topic = "temTopic", consumerGroup = "sixmonth",messageModel=MessageModel.CLUSTERING)
    public class RocketmqConsumerService implements RocketMQListener<String> {
    	
    	private Logger logger = LoggerFactory.getLogger(RocketmqConsumerService.class);
    
    /**	
     *  自动注入持久层
     *	@Autowired
     *	HeartRateDao heartRateDao;
     *
     */	
        @Override
        public void onMessage(String message) {
    		   if(message != null>0) {
    			   Long startTime = System.currentTimeMillis();
    			   System.out.println(message);
    			   try {
    				   JSONArray arr = JSONArray.parseArray(message);//解析json数据格式
    				   System.out.println("接收消息长度："+arr.size());
    				   Long endTime = System.currentTimeMillis();
    				   System.out.println("单次消费耗时："+(endTime-startTime)/1000);
    				   
    			   }catch(JSONException e) {
    				   logger.info("消费失败，json格式错误："+message);
    			   }
    		   
    	      }else {
    	    	logger.info("错误消息体："+message);
    	      }
        }
    }

*  注意事项：

1.官方代码需要开发工具安装lombok工具，同时项目引入lombok.jar才能运行代码；
    <dependency>
    	<groupId>org.projectlombok</groupId>
    	<artifactId>lombok</artifactId>
    	<scope>provided</scope>
    </dependency>
    
    2.如果需要将启动项starter打包成jar使用，需要使用maven-assembly-plugin插件进行依赖打包；
    <plugin>
    	<artifactId>maven-assembly-plugin</artifactId>
    	<configuration>
    		<descriptorRefs>
    			<descriptorRef>jar-with-dependencies</descriptorRef>
    		</descriptorRefs>
    		<archive>
    			<manifest>
    				<mainClass>com.sixmonth.rocketmq.RocketmqApplication</mainClass>
    			</manifest>
    		</archive>
    	</configuration>
    	<executions>
    		<execution>
    			<id>make-assembly</id>
    			<phase>package</phase>
    			<goals>
    				<goal>single</goal>
    			</goals>
    		</execution>
    	</executions>
    </plugin>     
    
    3.外部项目引入启动项jar包时，需在项目pom添加maven引入外部lib的配置，否则打包不成功；
    <plugin>
    	<artifactId>maven-compiler-plugin</artifactId>
    	
    	<configuration>
    		<source>1.8</source>
    		<target>1.8</target>
    		<encoding>UTF-8</encoding>
    		<compilerArguments>
    			<extdirs>${project.basedir}/lib</extdirs>
    		</compilerArguments>
    	</configuration>
    </plugin>
    
    4.引用启动类jar包时，需要在启动类添加扫描jar包的路径；
    @SpringBootApplication(scanBasePackages="com.sixmonth.rocketmq,org.apache.rocketmq.spring.starter")
    public class RocketmqProducerApplication {
    	public static void main(String[] args) {
    		SpringApplication.run(RocketmqProducerApplication.class, args);
    	}
    }
    
    5.集群消费如果是同一个消费组，则消费组内的消费实例平均分摊消费消息，如果是不同消费组，则会起到广播消费的效果；
    
    6.消费者注册监听有两种模式：并发消费(MessageListenerConcurrently)(默认)和有序消费(MessageListenerOrderly)模式；
      消费者消费消息有两种模式：广播消费(BROADCASTING)和集群消费(CLUSTERING)模式(默认)；
    
    7.有序消费一般需要和有序发送一起配合使用，有序发送通过MessageQueueSelector来指定消息发送到的MessageQueue，
    尽量让有序消息发送到同一个MessageQueue；
    
    8.开发参考文档api：https://help.aliyun.com/document_detail/29532.html?spm=a2c4g.11186623.6.541.26386c67Z88rLL；

### 结语 ###

1.  自行搭建rocketmq集群，也成为很多公司的使用阿里云消息队列的生产和消费方式；
2.  如果自己的服务器性能很不错的话，不妨使用这种方式；
3.  实践是检验认识真理性的唯一标准，不妨试一试~

[https_blog.csdn.net_alan_liuyue_article_details_86648280]: https://blog.csdn.net/alan_liuyue/article/details/86648280
[https_blog.csdn.net_alan_liuyue_article_details_86613896]: https://blog.csdn.net/alan_liuyue/article/details/86613896
[https_download.csdn.net_download_alan_liuyue_10937415]: https://download.csdn.net/download/alan_liuyue/10937415