diff --git a/README.md b/README.md index 6ff649a..5424e7a 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,10 @@ - github | Spring Boot 系列 ：[https://github.com/gf-huanchupk/SpringBootLearning](https://github.com/gf-huanchupk/SpringBootLearning),记得star哦！ - github | Spring Cloud 系列 ：[https://github.com/gf-huanchupk/SpringCloudLearning](https://github.com/gf-huanchupk/SpringCloudLearning),记得star哦！ +### RocketMQ 系列 +- [RocketMQ 教程 | RocketMQ 简介](http://blog.itwolfed.com/blog/98) +- [RocketMQ 教程 | RocketMQ 安装](http://blog.itwolfed.com/blog/99) +- [RocketMQ 教程 | RocketMQ 常用消息类型](http://blog.itwolfed.com/blog/101) ### Spring Boot 系列 @@ -84,7 +88,7 @@

- +
欢迎扫码或微信搜索公众号《程序员果果》关注我，更多精彩内容不能错过 ~~

diff --git a/springboot-rocketmq-message/.gitignore b/springboot-rocketmq-message/.gitignore new file mode 100644 index 0000000..1e4385b --- /dev/null +++ b/springboot-rocketmq-message/.gitignore @@ -0,0 +1,36 @@ +HELP.md +target/ +!.mvn/wrapper/maven-wrapper.jar +!**/src/main/**/target/ +!**/src/test/**/target/ + +### STS ### +.apt_generated +.classpath +.factorypath +.project +.settings +.springBeans +.sts4-cache + +### IntelliJ IDEA ### +.idea +*.iws +*.iml +*.ipr +mvnw.cmd +mvnw +.mvn + +### NetBeans ### +/nbproject/private/ +/nbbuild/ +/dist/ +/nbdist/ +/.nb-gradle/ +build/ +!**/src/main/**/build/ +!**/src/test/**/build/ + +### VS Code ### +.vscode/ diff --git a/springboot-rocketmq-message/pom.xml b/springboot-rocketmq-message/pom.xml new file mode 100644 index 0000000..310f5b8 --- /dev/null +++ b/springboot-rocketmq-message/pom.xml @@ -0,0 +1,57 @@ + + + 4.0.0 + + org.springframework.boot + spring-boot-starter-parent + 2.1.10.RELEASE + + + com.itwolfed + springboot-rocketmq-message + 0.0.1-SNAPSHOT + springboot-rocketmq-message + Demo project for Spring Boot + + 1.8 + + + + org.springframework.boot + spring-boot-starter-web + + + org.apache.rocketmq + rocketmq-spring-boot-starter + 2.1.1 + + + org.projectlombok + lombok + 1.18.16 + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + diff --git a/springboot-rocketmq-message/src/main/java/com/itwolfed/RocketmqMessageNormalApplication.java b/springboot-rocketmq-message/src/main/java/com/itwolfed/RocketmqMessageNormalApplication.java new file mode 100644 index 0000000..43e68eb --- /dev/null +++ b/springboot-rocketmq-message/src/main/java/com/itwolfed/RocketmqMessageNormalApplication.java @@ -0,0 +1,13 @@ +package com.itwolfed; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class RocketmqMessageNormalApplication { + + public static void main(String[] args) { + SpringApplication.run( RocketmqMessageNormalApplication.class, args ); + } + +} diff --git a/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/Consumer.java b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/Consumer.java new file mode 100644 index 0000000..090ac08 --- /dev/null +++ b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/Consumer.java @@ -0,0 +1,18 @@ +package com.itwolfed.msg; + +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; +import org.apache.rocketmq.spring.core.RocketMQListener; +import org.springframework.stereotype.Component; + +@Component +@Slf4j +@RocketMQMessageListener(topic = "Topic-Normal",consumerGroup = "consumer-msg-group") +public class Consumer implements RocketMQListener { + + @Override + public void onMessage(String message) { + log.info("Receive message：{}" , message); + } + +} diff --git a/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/DelayConsumer.java b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/DelayConsumer.java new file mode 100644 index 0000000..f85a4c7 --- /dev/null +++ b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/DelayConsumer.java @@ -0,0 +1,18 @@ +package com.itwolfed.msg; + +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; +import org.apache.rocketmq.spring.core.RocketMQListener; +import org.springframework.stereotype.Component; + +@Component +@Slf4j +@RocketMQMessageListener(topic = "Topic-Delay",consumerGroup = "consumer-delay-group") +public class DelayConsumer implements RocketMQListener { + + @Override + public void onMessage(String message) { + log.info("Receive message：{}" , message); + } + +} diff --git a/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/OrderConsumer.java b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/OrderConsumer.java new file mode 100644 index 0000000..994aa25 --- /dev/null +++ b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/OrderConsumer.java @@ -0,0 +1,18 @@ +package com.itwolfed.msg; + +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; +import org.apache.rocketmq.spring.core.RocketMQListener; +import org.springframework.stereotype.Component; + +@Component +@Slf4j +@RocketMQMessageListener(topic = "Topic-Order",consumerGroup = "consumer-order-group") +public class OrderConsumer implements RocketMQListener { + + @Override + public void onMessage(String message) { + log.info("Receive message：{}" , message); + } + +} diff --git a/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/Producer.java b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/Producer.java new file mode 100644 index 0000000..71b9709 --- /dev/null +++ b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/Producer.java @@ -0,0 +1,128 @@ +package com.itwolfed.msg; + +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.producer.LocalTransactionState; +import org.apache.rocketmq.client.producer.SendCallback; +import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.client.producer.SendStatus; +import org.apache.rocketmq.client.producer.TransactionSendResult; +import org.apache.rocketmq.common.message.Message; +import org.apache.rocketmq.spring.core.RocketMQTemplate; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.stereotype.Component; + +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +@Slf4j +@Component +public class Producer { + + + @Autowired + private RocketMQTemplate rocketMQTemplate; + + /** + * 普通消息 + */ + public void sendMsg() throws Exception { + + Message message = new Message( + // 普通消息所属的Topic + "Topic-Normal", + // Message Tag可理解为Gmail中的标签，对消息进行再归类，方便Consumer指定过滤条件在消息队列 RocketMQ 版的服务器过滤。 + "TagA", + // Message Body可以是任何二进制形式的数据。 + "Hello MQ".getBytes() + ); + rocketMQTemplate.getProducer().send( message ); + // 等同于上面的方式(常用) + //rocketMQTemplate.convertAndSend("Topic-Normal:TagA","Hello MQ".getBytes()); + } + + /** + * 异步发送普通消息 + */ + public void sendAsyncMsg() { + Map map = new HashMap<>(); + map.put( "name", "zs" ); + map.put( "age", 20 ); + + rocketMQTemplate.asyncSend( "Topic-Normal", map, new SendCallback() { + @Override + public void onSuccess(SendResult sendResult) { + // 消息发送成功。 + log.info( "async send success" ); + } + + @Override + public void onException(Throwable throwable) { + // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。 + log.info( "async send fail" ); + } + } ); + } + + /** + * 顺序消息 + */ + public void sendOrderlyMsg() { + //根据指定的hashKey按顺序发送 + for (int i = 0; i < 1000; i++) { + String orderId = "biz_" + i % 10; + // 分区顺序消息中区分不同分区的关键字段，Sharding Key与普通消息的key是完全不同的概念。 + // 全局顺序消息，该字段可以设置为任意非空字符串。 + String shardingKey = String.valueOf( orderId ); + try { + SendResult sendResult = rocketMQTemplate.syncSendOrderly( "Topic-Order", "send order msg".getBytes(), shardingKey ); + // 发送消息，只要不抛异常就是成功。 + if (sendResult != null) { + System.out.println( new Date() + " Send mq message success . msgId is:" + sendResult.getMsgId() ); + } + } catch (Exception e) { + // 消息发送失败，需要进行重试处理，可重新发送这条消息或持久化这条数据进行补偿处理。 + System.out.println( new Date() + " Send mq message failed" ); + e.printStackTrace(); + } + } + } + + /** + * 延时消息 + */ + public void sendDelayMsg() { + rocketMQTemplate.syncSend( "Topic-Delay", + MessageBuilder.withPayload( "Hello MQ".getBytes() ).build(), + 3000, + //设置延时等级3,这个消息将在10s之后发送(现在只支持固定的几个时间,详看delayTimeLevel) + //messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"; + 3 ); + } + + /** + * 批量消息 + */ + public void sendBatchMsg(List messages) { + rocketMQTemplate.syncSend( "springboot-rocketmq", messages ); + + } + + /** + * 事务消息 + */ + public void sendTransactionMsg(){ + TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction( + "Topic-Tx:TagA", + MessageBuilder.withPayload( "Hello MQ transaction===".getBytes() ).build(), + null ); + SendStatus sendStatus = transactionSendResult.getSendStatus(); + LocalTransactionState localTransactionState = transactionSendResult.getLocalTransactionState(); + + System.out.println( new Date() + " Send mq message status "+ sendStatus +" , localTransactionState "+ localTransactionState ); + } + + +} diff --git a/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/TxConsumerListener.java b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/TxConsumerListener.java new file mode 100644 index 0000000..84975de --- /dev/null +++ b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/TxConsumerListener.java @@ -0,0 +1,22 @@ +package com.itwolfed.msg; + +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; +import org.apache.rocketmq.spring.core.RocketMQListener; +import org.springframework.stereotype.Component; + + +@Component +@Slf4j +@RocketMQMessageListener(topic = "Topic-Tx",consumerGroup = "consumer-tx-group") +public class TxConsumerListener implements RocketMQListener { + + + + @Override + public void onMessage(String message) { + + log.info("Receive message：{}" , message); + } + +} diff --git a/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/TxProducerListener.java b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/TxProducerListener.java new file mode 100644 index 0000000..3eb5c8a --- /dev/null +++ b/springboot-rocketmq-message/src/main/java/com/itwolfed/msg/TxProducerListener.java @@ -0,0 +1,45 @@ +package com.itwolfed.msg; + +import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener; +import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener; +import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState; +import org.springframework.messaging.Message; +import org.springframework.stereotype.Component; + + +@Component +@RocketMQTransactionListener +public class TxProducerListener implements RocketMQLocalTransactionListener { + + @Override + public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) { + // 执行本地事务 + System.out.println("TX message listener execute local transaction"); + RocketMQLocalTransactionState result; + try { + // 业务代码（ 例如下订单 ） + result = RocketMQLocalTransactionState.COMMIT; + } catch (Exception e) { + System.out.println("execute local transaction error"); + result = RocketMQLocalTransactionState.UNKNOWN; + } + return result; + } + + @Override + public RocketMQLocalTransactionState checkLocalTransaction(Message msg) { + // 检查本地事务（ 例如检查下订单是否成功 ） + System.out.println("TX message listener check local transaction"); + RocketMQLocalTransactionState result; + try { + //业务代码（ 根据检查结果，决定是COMMIT或ROLLBACK ） + result = RocketMQLocalTransactionState.COMMIT; + } catch (Exception e) { + // 异常就回滚 + System.out.println("check local transaction error"); + result = RocketMQLocalTransactionState.ROLLBACK; + } + return result; + } + +} diff --git a/springboot-rocketmq-message/src/main/resources/application.yml b/springboot-rocketmq-message/src/main/resources/application.yml new file mode 100644 index 0000000..fa3511d --- /dev/null +++ b/springboot-rocketmq-message/src/main/resources/application.yml @@ -0,0 +1,7 @@ +server: + port: 8080 + +rocketmq: + name-server: 172.16.250.129:9876 + producer: + group: myGroup \ No newline at end of file diff --git a/springboot-rocketmq-message/src/test/java/com/itwolfed/RocketmqMessageNormalApplicationTests.java b/springboot-rocketmq-message/src/test/java/com/itwolfed/RocketmqMessageNormalApplicationTests.java new file mode 100644 index 0000000..29f75c4 --- /dev/null +++ b/springboot-rocketmq-message/src/test/java/com/itwolfed/RocketmqMessageNormalApplicationTests.java @@ -0,0 +1,62 @@ +package com.itwolfed; + + +import com.itwolfed.msg.Producer; +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.test.context.junit4.SpringRunner; + + +@RunWith(SpringRunner.class) +@SpringBootTest +public class RocketmqMessageNormalApplicationTests { + + @Autowired + private Producer producer; + + + + /** + * 测试普通消息 + */ + @Test + public void sendMsg() throws Exception { + producer.sendMsg(); + } + + /** + * 异步发送普通消息 + */ + @Test + public void sendAsyncMsg() { + producer.sendAsyncMsg(); + } + + /** + * 顺序消息 + */ + @Test + public void sendOrderlyMsg() { + producer.sendOrderlyMsg(); + } + + /** + * 延时消息 + */ + @Test + public void sendDelayMsg() { + producer.sendDelayMsg(); + } + + /** + * 事务消息 + */ + @Test + public void sendTransactionMsg(){ + producer.sendTransactionMsg(); + } + + +}