网站自建设需要买什么时候开始,长沙官网优化推广,国贸行业的网站建设,室内设计高端网站RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件#xff0c;前身是MetaQ#xff0c;是阿里参考Kafka特点研发的一个队列模型的消息中间件#xff0c;后开源给apache基金会成为了apache的顶级开源项目#xff0c;具有高性能、高可靠、高实时、分布式特点。
环境搭…RocketMQ是一个纯Java、分布式、队列模型的开源消息中间件前身是MetaQ是阿里参考Kafka特点研发的一个队列模型的消息中间件后开源给apache基金会成为了apache的顶级开源项目具有高性能、高可靠、高实时、分布式特点。
环境搭建
采用docker-compose搭建具体配置如下 version: 3
services:# rocket mq name serverrmqnamesrv:image: apache/rocketmq:4.9.6restart: alwayscontainer_name: rocket-server# environment:# JAVA_OPT_EXT: -server -Xms64m -Xmx64m -Xmn64m# volumes:# 映射本地目录权限一定要设置为 777 权限否则启动不成功# - ../volumes/data/rocket/server/logs:/home/rocketmq/logsnetworks:- rocketmqports:- 9876:9876command: sh mqnamesrv# rocket mq brokerrmqbroker:image: apache/rocketmq:4.9.6restart: alwayscontainer_name: rocket-brokervolumes:# 映射本地目录权限一定要设置为 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.confenvironment:- NAMESRV_ADDRrmqnamesrv:9876# - JAVA_OPTS:-Duser.home/opt- JAVA_OPT_EXT-server -Xms64m -Xmx64m -Xmn64mdepends_on:- rmqnamesrvnetworks:- rocketmqports:- 10909:10909- 10911:10911command: sh mqbroker -c /opt/rocketmq-4.9.6/conf/broker.conf# rocket console 这个可以不需要rmqdashboard:image: apacherocketmq/rocketmq-dashboard:1.0.0restart: alwayscontainer_name: rocket-dashboardenvironment:- JAVA_OPTS-Drocketmq.config.namesrvAddrrmqnamesrv:9876 -Dserver.port8180 -Drocketmq.config.isVIPChannelfalse# - JAVA_OPT_EXT-Xms128m -Xmx128m -Xmn128mdepends_on:- rmqnamesrvnetworks:- rocketmqports:- 8180:8180networks:rocketmq:driver: bridgestack: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 version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsdparentartifactIdspringboot-demo/artifactIdgroupIdcom.et/groupIdversion1.0-SNAPSHOT/version/parentmodelVersion4.0.0/modelVersionartifactIdrocketmq/artifactIdpropertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.target/propertiesdependenciesdependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-spring-boot-starter/artifactIdversion2.1.1/version/dependencydependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-client/artifactIdversion4.8.0/version/dependencydependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-common/artifactIdversion4.8.0/version/dependencydependencygroupIdjunit/groupIdartifactIdjunit/artifactIdscopetest/scope/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.springframework/groupIdartifactIdspring-test/artifactIdscopetest/scope/dependency/dependencies/project二、配置文件和启动类 server:port: 8088
#rocketmq配置
rocketmq:name-server: 10.11.68.77:9876# 生产者配置producer:isOnOff: on# 发送同一类消息的设置为同一个group保证唯一group: hyh-rocketmq-groupgroupName: hyh-rocketmq-group# 服务地址namesrvAddr: 10.11.68.77:9876# 消息最大长度 默认1024*4(4M)maxMessageSize: 4096# 发送消息超时时间,默认3000sendMsgTimeout: 3000# 发送消息失败重试次数默认2retryTimesWhenSendFailed: 2SpringBootApplication
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 RocketMQListenerPerson{Overridepublic 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);Autowiredprivate RocketMQTemplate rocketMQTemplate;PostConstructpublic 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() {Overridepublic void onSuccess(SendResult sendResult) {LOG.info(---发送MQ成功---);}Overridepublic void onException(Throwable throwable) {throwable.printStackTrace();LOG.error(---发送MQ失败---throwable.getMessage(), throwable.getMessage());}};}PreDestroypublic void destroy() {LOG.info(---RocketMq助手注销---);}}
六、测试
RunWith(SpringRunner.class)
SpringBootTest
public class MQTests {Autowiredprivate RocketMqHelper rocketMqHelper;Testpublic void testProducter() throws InterruptedException {for(int i0;i1000000;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
