当前位置：首页 > 谷歌SEO >

2021年新开传奇盛世网页游戏开服时间表是什么？

96SEO 2026-02-20 07:52 0

一般rocketmq使用在微服务项目中#xff0c;属于分模块使用。

这里使用springboot单体项目来模拟使用。

下载rocketmq和启动#xff1a;

一般rocketmq使用在微服务项目中属于分模块使用。

这里使用springboot单体项目来模拟使用。

下载rocketmq和启动

(zhihu.com)https://zhuanlan.zhihu.com/p/644944370

一、创建springboot项目

dependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scope/dependencydependencygroupIdorg.apache.rocketmq/groupIdartifactIdrocketmq-spring-boot-starter/artifactIdversion2.0.4/version/dependency!--

还有其它需要的jar包自由引入注fastjson不要使用低于1.2.60版本会有安全漏洞

--dependencygroupIdcom.alibaba/groupIdartifactIdfastjson/artifactIdversion1.2.62/version/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId/dependency/dependencies

三、代码实现

什么是生产者就比喻个简单的例子。

比如我们要新增用户那么这个新增保存动作可以认为是生产者他产生了数据要将数据保存进数据库。

消费者

就比喻个简单的例子。

用户在新增的时候他会调用接口用于保存到数据库那么处理这个数据的方法你可以理解为消费者。

不过在mq中生产者是将消息发送到mq服务队列中会根据主题Topic的不同发往不同的频道。

而消费者只需要监听这个Topic主题即可。

只要这个topic有消息来了那么消费者就会进行消费。

后面代码里有详细的注释告知大家如何使用生产者和消费者。

Tomcat

消息发送超时时长默认3sretry-times-when-send-failed:

同步发送消息失败重试次数默认2retry-times-when-send-async-failed:

异步发送消息失败重试次数默认2

com.example.rocketmqdemo.model.User;

import

org.apache.rocketmq.client.producer.SendCallback;

import

org.apache.rocketmq.client.producer.SendResult;

import

org.apache.rocketmq.spring.core.RocketMQTemplate;

import

org.springframework.beans.factory.annotation.Autowired;

import

org.springframework.beans.factory.annotation.Value;

import

org.springframework.messaging.support.MessageBuilder;

import

org.springframework.stereotype.Component;

import

org.springframework.util.CollectionUtils;import

java.util.List;Slf4j

{Value(${rocketmq.producer.send-message-timeout})private

Integer

直接注入使用用于发送消息到broker服务器Autowiredprivate

RocketMQTemplate

普通发送这里的参数对象User可以随意定义可以发送个对象也可以是字符串等*/public

void

{rocketMQTemplate.convertAndSend(topic

:tag1,

MessageBuilder.withPayload(user).build());

等价于上面一行}/***

发送同步消息阻塞当前线程等待broker响应发送结果这样不太容易丢失消息*

msgBody也可以是对象sendResult为返回的发送结果*/public

SendResult

rocketMQTemplate.syncSend(topic,

MessageBuilder.withPayload(msgBody).build());log.info(【sendMsg】sendResult{},

JSON.toJSONString(sendResult));return

sendResult;}/***

发送异步消息通过线程池执行发送到broker的消息任务执行完后回调在SendCallback中可处理相关成功失败时的逻辑*

void

{rocketMQTemplate.asyncSend(topic,

MessageBuilder.withPayload(msgBody).build(),

new

处理消息发送成功逻辑log.info(【sendMsg】sendResult{},

JSON.toJSONString(sendResult));}Overridepublic

void

处理消息发送异常逻辑log.info(【sendMsg】sendResult{},

发送异常

throwable.getMessage());}});}/***

发送延时消息上面的发送同步消息delayLevel的值就为0因为不延时*

在start版本中

{rocketMQTemplate.syncSend(topic,

MessageBuilder.withPayload(msgBody).build(),

messageTimeOut,

发送单向消息只负责发送消息不等待应答不关心发送结果如日志*/public

void

{rocketMQTemplate.sendOneWay(topic,

MessageBuilder.withPayload(msgBody).build());}/***

发送带tag的消息直接在topic后面加上:tag*/public

SendResult

rocketMQTemplate.syncSend(topic

:tag2,

MessageBuilder.withPayload(msgBody).build());}/****

服务生产者顺序消息*

{//顺序消息//选择器规则构建rocketMQTemplate.setMessageQueueSelector((list,

message,

(!CollectionUtils.isEmpty(users))

{for

{MessageBuilder.withPayload(users.toString()).build();rocketMQTemplate.sendOneWayOrderly(topic:sendFIFOMsg,

user,

String.valueOf(user.getId()));}}}

1.5消费者服务

com.example.rocketmqdemo.model.User;

import

org.apache.rocketmq.common.message.MessageExt;

import

org.apache.rocketmq.spring.annotation.ConsumeMode;

import

org.apache.rocketmq.spring.annotation.RocketMQMessageListener;

import

org.apache.rocketmq.spring.core.RocketMQListener;

import

org.springframework.stereotype.Component;

import

org.springframework.stereotype.Service;Slf4j

Component

topic需要和生产者的topic一致consumerGroup属性是必须指定的内容可以随意//

selectorExpression的意思指的就是tag默认为“*”不设置的话会监听所有消息ServiceRocketMQMessageListener(topic

RLT_TEST_TOPIC,

注意这个ConsumerSend2和上面ConsumerSend在没有添加tag做区分时不能共存//

不然生产者发送一条消息这两个都会去消费如果类型不同会有一个报错所以实际运用中最好加上tag写这只是让你看知道就行ServiceRocketMQMessageListener(topic

RLT_TEST_TOPIC,

Con_Group_Two,selectorExpression

xxx)public

{log.info(ConsumerSend2监听到消息str{},

str);}}//

MessageExt是一个消息接收通配符不管发送的是String还是对象都可接收当然也可以像上面明确指定类型我建议还是指定类型较方便ServiceRocketMQMessageListener(topic

RLT_TEST_TOPIC,

String(body);log.info(tag2监听到消息msg{},

msg);}}/***

顺序消费*/ServiceRocketMQMessageListener(consumerGroup

Orderly-Consumer,

RLT_TEST_TOPIC,selectorExpression

sendFIFOMsg,

{System.out.println(线程Thread.currentThread()内容为:

new

String(message.getBody())队列序号:message.getQueueId(),消息msgId:message.getMsgId());}}

1.6User实体类

com.example.rocketmqdemo.model.User;

import

com.example.rocketmqdemo.producer.MQProducerService;

import

org.apache.rocketmq.client.producer.SendResult;

import

org.springframework.beans.factory.annotation.Autowired;

import

org.springframework.http.ResponseEntity;

import

org.springframework.web.bind.annotation.GetMapping;

import

org.springframework.web.bind.annotation.RequestMapping;

import

org.springframework.web.bind.annotation.RestController;import

java.util.ArrayList;

mqProducerService;GetMapping(/send)public

void

User();user.setAge(28);user.setName(曹震);user.setSex(男);mqProducerService.send(user);}GetMapping(/sendTag)public

sendTag()

mqProducerService.sendTagMsg(带有tag的字符消息);return

ResponseEntity.ok(sendResult);}GetMapping(/sendMsg)public

sendMsg()

mqProducerService.sendMsg(曹震测试);return

ResponseEntity.ok(sendResult);}GetMapping(/sendFIFOMsg)public

void

User();user.setId(1);user.setSex(男);user.setName(曹震);user.setAge(28);user.setDesc(创建订单);users.add(user);User

user1

User();user1.setId(2);user1.setSex(男);user1.setName(贾耀旗);user1.setAge(25);user1.setDesc(创建订单);users.add(user1);User

user2

User();user2.setId(1);user2.setSex(男);user2.setName(曹震);user2.setAge(28);user2.setDesc(订单付款);users.add(user2);User

user3

User();user3.setId(1);user3.setSex(男);user3.setName(曹震);user3.setAge(28);user3.setDesc(订单完成);users.add(user3);User

user4

User();user4.setId(1);user4.setSex(男);user4.setName(曹震);user4.setAge(28);user4.setDesc(订单推送);users.add(user4);User

user5

User();user5.setId(2);user5.setSex(男);user5.setName(贾耀旗);user5.setAge(25);user5.setDesc(订单付款);users.add(user5);User

user6

User();user6.setId(2);user6.setSex(男);user6.setName(贾耀旗);user6.setAge(25);user6.setDesc(订单完成);users.add(user6);mqProducerService.sendFIFOMsg(users);}}

1.8我们来启动服务看下效果

这个方法进行测试。

可以看出这里的代码其实是内容顺序是乱的我们先看调用成功后的结果

线程Thread[ConsumeMessageThread_3,5,main]内容为:{id:1,name:曹震,age:28,sex:男,desc:创建订单}队列序号:1,消息msgId:A9FE29E30E3800DAD5DC7F03A485001C