一、rocketmq介绍
RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件，前身是MetaQ，是阿里参考Kafka特点研发的一个队列模型的消息中间件，后开源给apache基金会成为了apache的顶级开源项目，具有高性能、高可靠、高实时、分布式特点。
二、rocketmq环境搭建
采用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
三、代码工程
pom.xml
<?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>entity
package com.et59.rocketmq.entity;
/**
* @author liuhaihua
* @version 1.0
* @ClassName Person
* @Description todo
* @date 2023年12月29日 9:52
*/
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
package com.et59.rocketmq.listener;
import com.et59.rocketmq.entity.Person;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;
/**
* @Author: heyuhua
* @Date: 2021/1/8 15:55
*/
@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}", topic = "PERSON_ADD")
public class PersonMqListener implements RocketMQListener<Person> {
@Override
public void onMessage(Person person) {
System.out.println("接收到消息，开始消费..name:" + person.getName() + ",age:" + person.getAge());
}
}util
package com.et59.rocketmq.util;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
/**
* RocketMq助手
*
* @Author: heyuhua
* @Date: 2020/1/8 10:03
*/
@Component
public class RocketMqHelper {
/**
* 日志
*/
private static final Logger LOG = LoggerFactory.getLogger(RocketMqHelper.class);
/**
* rocketmq模板注入
*/
@Autowired
private RocketMQTemplate rocketMQTemplate;
@PostConstruct
public void init() {
LOG.info("---RocketMq助手初始化---");
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
*/
public void asyncSend(Enum topic, Message<?> message) {
asyncSend(topic.name(), message, getDefaultSendCallBack());
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
* @param sendCallback 回调函数
*/
public void asyncSend(Enum topic, Message<?> message, SendCallback sendCallback) {
asyncSend(topic.name(), message, sendCallback);
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
*/
public void asyncSend(String topic, Message<?> message) {
rocketMQTemplate.asyncSend(topic, message, getDefaultSendCallBack());
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
* @param sendCallback 回调函数
*/
public void asyncSend(String topic, Message<?> message, SendCallback sendCallback) {
rocketMQTemplate.asyncSend(topic, message, sendCallback);
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
* @param sendCallback 回调函数
* @param timeout 超时时间
*/
public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout) {
rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout);
}
/**
* 发送异步消息
*
* @param topic 消息Topic
* @param message 消息实体
* @param sendCallback 回调函数
* @param timeout 超时时间
* @param delayLevel 延迟消息的级别
*/
public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout, int delayLevel) {
rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout, delayLevel);
}
/**
* 发送顺序消息
*
* @param message
* @param topic
* @param hashKey
*/
public void syncSendOrderly(Enum topic, Message<?> message, String hashKey) {
syncSendOrderly(topic.name(), message, hashKey);
}
/**
* 发送顺序消息
*
* @param message
* @param topic
* @param hashKey
*/
public void syncSendOrderly(String topic, Message<?> message, String hashKey) {
LOG.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey);
rocketMQTemplate.syncSendOrderly(topic, message, hashKey);
}
/**
* 发送顺序消息
*
* @param message
* @param topic
* @param hashKey
* @param timeout
*/
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函数
*
* @return
*/
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助手注销---");
}
}配置文件
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启动类
package com.et59.rocketmq;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class DemoApplication {
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
}code repository
https://github.com/Harries/springboot-demo
四、测试
package com.et59.rocketmq;
import com.et59.rocketmq.entity.Person;
import com.et59.rocketmq.util.RocketMqHelper;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.test.context.junit4.SpringRunner;
@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五、引用
https://rocketmq.apache.org/
http://www.liuhaihua.cn/archives/710205.html
