SpringBoot集成RocketMQ

RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里参考Kafka特点研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点。

环境搭建：

采用docker-compose搭建，具体配置如下

version: '3'
services:
  # rocket mq name server
  rmqnamesrv:
    image: apache/rocketmq:4.9.6
    restart: always
    container_name: rocket-server
      # environment:
      #   JAVA_OPT_EXT: "-server -Xms64m -Xmx64m -Xmn64m"
      # volumes:
      # 映射本地目录权限一定要设置为 777 权限，否则启动不成功
    # - ../volumes/data/rocket/server/logs:/home/rocketmq/logs
    networks:
      - rocketmq
    ports:
      - 9876:9876
    command: sh mqnamesrv
  # rocket mq broker
  rmqbroker:
    image: apache/rocketmq:4.9.6
    restart: always
    container_name: rocket-broker
    volumes:
      # 映射本地目录权限一定要设置为 777 权限，否则启动不成功
      # - ../volumes/data/rocket/broker/logs:/home/rocketmq/logs
      # - ../volumes/data/rocket/broker/store:/home/rocketmq/store
      - ./config/broker.conf:/opt/rocketmq-4.9.6/conf/broker.conf
    environment:
      - NAMESRV_ADDR=rmqnamesrv:9876
      # - JAVA_OPTS:=-Duser.home=/opt
      - JAVA_OPT_EXT=-server -Xms64m -Xmx64m -Xmn64m
    depends_on:
      - rmqnamesrv
    networks:
      - rocketmq
    ports:
      - 10909:10909
      - 10911:10911
    command: sh mqbroker -c /opt/rocketmq-4.9.6/conf/broker.conf
  # rocket console 这个可以不需要
  rmqdashboard:
    image: apacherocketmq/rocketmq-dashboard:1.0.0
    restart: always
    container_name: rocket-dashboard
    environment:
      - JAVA_OPTS=-Drocketmq.config.namesrvAddr=rmqnamesrv:9876 -Dserver.port=8180 -Drocketmq.config.isVIPChannel=false
      # - JAVA_OPT_EXT=-Xms128m -Xmx128m -Xmn128m
    depends_on:
      - rmqnamesrv
    networks:
      - rocketmq
    ports:
      - 8180:8180

networks:
  rocketmq:
    driver: bridge
  stack:
    driver: bridge

运行docker-compose：

docker-compose -f docker-compose-rocketmq.yml -p rocketmq up -d
注：修改 xx/rocketmq/rocketmq_broker/conf/broker.conf中配置brokerIP1为宿主机IP

访问地址：http://ip地址:8180：

一、依赖

<?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">
    <parent>
        <artifactId>springboot-demo</artifactId>
        <groupId>com.et</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>rocketmq</artifactId>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-spring-boot-starter</artifactId>
            <version>2.1.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.8.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-common</artifactId>
            <version>4.8.0</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-test</artifactId>
            <scope>test</scope>
        </dependency>

    </dependencies>

</project>

二、配置文件和启动类

server:
  port: 8088
#rocketmq配置
rocketmq:
  name-server: 10.11.68.77:9876
  # 生产者配置
  producer:
    isOnOff: on
    # 发送同一类消息的设置为同一个group，保证唯一
    group: hyh-rocketmq-group
    groupName: hyh-rocketmq-group
    # 服务地址
    namesrvAddr: 10.11.68.77:9876
    # 消息最大长度 默认1024*4(4M)
    maxMessageSize: 4096
    # 发送消息超时时间,默认3000
    sendMsgTimeout: 3000
    # 发送消息失败重试次数，默认2
    retryTimesWhenSendFailed: 2

@SpringBootApplication
public class DemoApplication {

   public static void main(String[] args) {
      SpringApplication.run(DemoApplication.class, args);
   }
}

三、实体类

public class Person {
    private String name;
    private int age;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public int getAge() {
        return age;
    }

    public void setAge(int age) {
        this.age = age;
    }
}

四、listener监听

@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}", topic = "PERSON_ADD") //配置文件中：hyh-rocketmq-group
public class PersonMqListener implements RocketMQListener<Person>{
	
	@Override
	public void onMessage(Person person){
		System.out.println("接收到消息，开始消费..name:" + person.getName() + ",age:" + person.getAge());
	}
}

五、工具util

@Component
public class RocketMqHelper{
	
	private static final Logger LOG = LoggerFactory.getLogger(RocketMqHelper.class);

	@Autowired
	private RocketMQTemplate rocketMQTemplate;

	@PostConstruct
	public void init(){
		LOG.info("---RocketMq助手初始化---");
	}

	//发送异步消息
	public void asyncSend(Enum topic,Message<?> message){
		asyncSend(topic.name(), message, getDefaultSendCallBack());
	}

	public void asyncSend(Enum topic, Message<?> message, SendCallback sendCallback) {
        asyncSend(topic.name(), message, sendCallback);
    }

	public void asyncSend(String topic, Message<?> message) {
        rocketMQTemplate.asyncSend(topic, message, getDefaultSendCallBack());
    }

	public void asyncSend(String topic, Message<?> message, SendCallback sendCallback) {
        rocketMQTemplate.asyncSend(topic, message, sendCallback);
    }

	public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout) {
        rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout);
    }

	 public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout, int delayLevel) {
        rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout, delayLevel);
    }

	//发送给顺序消息
	public void syncSendOrderly(Enum topic, Message<?> message, String hashKey) {
        syncSendOrderly(topic.name(), message, hashKey);
    }
	
	public void syncSendOrderly(String topic, Message<?> message, String hashKey) {
        LOG.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey);
        rocketMQTemplate.syncSendOrderly(topic, message, hashKey);
    }

	public void syncSendOrderly(String topic, Message<?> message, String hashKey, long timeout) {
        LOG.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey + ",timeout:" + timeout);
        rocketMQTemplate.syncSendOrderly(topic, message, hashKey, timeout);
    }

	//默认Callback函数
	private SendCallback getDefaultSendCallBack() {
        return new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                LOG.info("---发送MQ成功---");
            }

            @Override
            public void onException(Throwable throwable) {
                throwable.printStackTrace();
                LOG.error("---发送MQ失败---"+throwable.getMessage(), throwable.getMessage());
            }
        };
    }

	@PreDestroy
    public void destroy() {
        LOG.info("---RocketMq助手注销---");
    }

}

六、测试

@RunWith(SpringRunner.class)
@SpringBootTest
public class MQTests {
   @Autowired
   private RocketMqHelper rocketMqHelper;

   @Test
   public void testProducter() throws InterruptedException {
      for(int i=0;i<1000000;i++){
         Person person = new Person();
         person.setName("heyuhua");
         person.setAge(25);
         rocketMqHelper.asyncSend("PERSON_ADD", MessageBuilder.withPayload(person).build());
         Thread.sleep(1000);
      }
   }

}

结果：

2024-02-02 15:20:27.101 INFO 11188 --- [ublicExecutor_3] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:28.116 INFO 11188 --- [ublicExecutor_4] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:29.130 INFO 11188 --- [ublicExecutor_5] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:30.131 INFO 11188 --- [ublicExecutor_7] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:31.142 INFO 11188 --- [ublicExecutor_8] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:32.156 INFO 11188 --- [ublicExecutor_9] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
2024-02-02 15:20:33.167 INFO 11188 --- [blicExecutor_10] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
2024-02-02 15:20:34.182 INFO 11188 --- [ublicExecutor_3] com.et59.rocketmq.util.RocketMqHelper : ---发送MQ成功---
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25
接收到消息，开始消费..name:heyuhua,age:25