当前位置: 首页 > news >正文

石家庄市市政建设工程公司网站今日国内新闻头条

石家庄市市政建设工程公司网站,今日国内新闻头条,如何做网站活动,网站建设服务费是否无形资产文章目录 前言基本概念消息和主题相关发送普通消息 发送顺序消息RocketMQTemplate的API介绍参考资料: 前言 本文主要有以下内容: 简单消息的发送顺序消息的发送RocketMQTemplate的API介绍 环境搭建: RocketMQ的安装教程:在官网…

文章目录

  • 前言
  • 基本概念
    • 消息和主题相关
    • 发送普通消息
  • 发送顺序消息
  • RocketMQTemplate的API介绍
  • 参考资料:

前言

本文主要有以下内容:

  • 简单消息的发送
  • 顺序消息的发送
  • RocketMQTemplate的API介绍

环境搭建:
RocketMQ的安装教程:在官网上下载bin文件,解压到本地,并配置环境变量,如下图所示:
在这里插入图片描述

在 Spring boot 项目中引入 RocketMQ 依赖:

<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.3</version>
</dependency>

在application.yml增加相关配置:

server:port: 10001
rocketmq:name-server: 127.0.0.1:9876producer:group: springboot_produce_group # 必须指定groupsend-message-timeout: 3000 # 消息发送超时时长,默认3sretry-times-when-send-failed: 3 # 同步发送消息失败重试次数,默认2retry-times-when-send-async-failed: 3 # 异步发送消息失败重试次数,默认2consumer:group: springboot_consumer_group

在 Spring Boot 中使用RocketMQ很简单直接注入RocketMQTemplate对象即可:

@Resource
private RocketMQTemplate rocketMQTemplate;

基本概念

消息和主题相关

消息 message:通信交互的载体,分为事务消息,半事务消息,延迟消息,顺序消息等。
主题 topic:一类消息的集合,逻辑概念。
队列 queue:主题由一个队列或者多个队列构成,当消息发送到某一个主题时,需要选择某一个队列。
偏移量 offset:消息追加到主题的队列后会分配一个数值,表示该队列的几条消息。
消费者相关:
消费组 consume group:消费组用于订阅主题消费消息,可以订阅多个主题,一个消费组可以有多个消费者。
广播模式:同一个消费组内的所有消费者都会消费订阅主题的所有消息。即一条消息会被该消费者组的所有消费者消费。
集群模式:同一个消费组内的所有消费者只消费订阅主题的一部分消息,即一条消息只会被改消费组的一个消费者消费。
并发消费:同一个队列的消息由多线程消费且不保证消息的顺序。
顺序消费:保证同一队列的消息按顺序消费。

发送普通消息

创建MsgController,代码如下:

@RestController
@RequestMapping("send/")
@CrossOrigin(allowedHeaders = "*", origins = "*")
@Slf4j
public class MsgController {@Resourceprivate RocketMQTemplate rocketMQTemplate;@GetMapping("normal")public void sendNormalMsg() {Message<String> msg = MessageBuilder.withPayload("Hello,RocketMQ Normal_msg").build();rocketMQTemplate.send("normal_msg", msg);}
}

创建消息的消费者,只需要实现RocketMQListener接口中的方法即可,代码如下:

@Component
@RocketMQMessageListener(topic = "normal_msg", consumerGroup = "consumer_normal")
@Slf4j
public class NormalMsgConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {log.info("Receive Normal Msg: {}",message);}
}

@RocketMQMessageListener注解用在消费者类上,指定当前类消费的主题。

topic:指定消费者的主题 comsumerGroup:指定消费者组(Consumer Group)名称,用于区分不同的消费者。

启动项目,运行结果如下图所示:
在这里插入图片描述

发送顺序消息

顺序消息:保证同一队列的消息按顺序消费。
在MsgController 中添加如下代码:

@GetMapping("order")
public void sendOrderMsg(){
​log.info("开始发送顺序消息");for (int j = 0; j < 10; j++) {Message<String> sendOrderMsg = MessageBuilder.withPayload("Send Order Msg = " + j + " time: "+ LocalDateTime.now()).build();try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}rocketMQTemplate.convertAndSend("msg:order", sendOrderMsg);}log.info("顺序消息发送结束");
}

创建对应topic消息的消费者,代码如下所示:

@Component
@RocketMQMessageListener(topic = "msg",consumerGroup = "consumer_order_group",selectorExpression = "order",messageModel = MessageModel.CLUSTERING,selectorType = SelectorType.TAG)
@Slf4j
public class OrderMsgConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String message) {log.info("Receive Order Msg: {}",message);}
}

@RocketMQMessageListener其他属性介绍:

  • selectorExpression: 消息选择表达式,用于过滤消息,只有满足表达式条件的消息才会被消费。默认值为 *,表示订阅所有消息。

全匹配:*,默认值。
属性匹配:指定tag = ‘tagName’,上面的代码就可以改写为"tag = ‘order’"
表达式匹配:需要指定selectType = SelectorType.SQL92,见下面。

  • selectorType:指明了消息选择通过tag的方式,默认值SelectorType.TAG。可选值有SelectorType.SQL92

TAG:支持"tagName"的方式配置,如果有多个标签则用||进行连接
SQL92:关键字有AND, OR, NOT, BETWEEN, IN, TRUE, FALSE, IS, NULL。支持的数据类型有Boolean, String, Decimal, Float number等。使用方式如(a > 10 AND a < 100) OR (b IS NOT NULL AND b=TRUE)

  • messageModel:消息模式,可选值为 MessageModel.CLUSTERING(默认)或 MessageModel.BROADCASTING,分别表示集群模式和广播模式。

重新启动项目,运行结果如下图所示:
在这里插入图片描述

RocketMQTemplate的API介绍

在上面的api使用中,都没有去关注是否消息发送的状态,如是否成功,发送到了哪一个队列等。接下来就介绍一下相关API的使用

带返回值的发送普通消息SendResult syncSend(String destination, Message<?> message);

在MsgController添加如下代码:

@GetMapping("normal_result")
public void sendNormalResultMsg() {Message<String> msg = MessageBuilder.withPayload("normal_return_result").build();SendResult normalMsg = rocketMQTemplate.syncSend("normal_msg", msg);log.info("normalMsg = {}",normalMsg);
}

在这里插入图片描述

如log所示,可以看到发送状态等信息。

发送异步消息,在MsgController中添加如下代码:

@GetMapping("callback")
public void sendNormalResultMsgWithCallback(){Message<String> msg = MessageBuilder.withPayload("normal_return_result").build();rocketMQTemplate.asyncSend("normal_msg", msg, new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {log.info("success");}@Overridepublic void onException(Throwable throwable) {log.info("error");}});
}

运行结果如下所示:
在这里插入图片描述

发送顺序消息:在第二部分以及展示过了也可以用如下代码替换

rocketMQTemplate.convertAndSend("msg:order", sendOrderMsg);
// 替换为
rocketMQTemplate.syncSendOrderly("msg:order", sendOrderMsg,String.valueOf(j));

发送单向消息

@GetMapping("oneway")
public void  sendOneWay(){Message<String> oneWay = MessageBuilder.withPayload("Send Order Msg = " + " time: "+ LocalDateTime.now()).build();rocketMQTemplate.sendOneWay("normal_msg",oneWay);
}

运行结果如下图所示:
在这里插入图片描述

发送事务消息:暂不举例,后续补充
发送事务消息带回调:和syncSend()类似,后续补充相关用法。

参考资料:

  • 《RocketMQ 实战》
http://www.shuangfujiaoyu.com/news/18573.html

相关文章:

  • 英文医疗网站建设手机优化是什么意思
  • 深圳如何做网站百度网首页登录入口
  • 素材网站的下载服务器怎么做seo营销优化
  • 不属于c2c网站的是怎么样建立自己的网站
  • 公司做外地网站营销型网站建设目标
  • 网站做响应式还是移动端网站手机优化
  • 个人网站制作模板电商培训内容
  • ecshop模板网seo综合查询爱站
  • 专注徐州网站建设百度快照优化培训班
  • 可以做渐变色块拼接的网站青岛做网站推广
  • 建设网站费用网址收录网站
  • 代充网站怎么做刷僵尸粉网站推广
  • 网站建设服务开发网络培训心得体会总结
  • 誉重网站建设公司高级搜索入口
  • 网站的三大因素软文类型
  • 珠海市城市建设档案馆网站百度今日小说排行榜
  • 物流公司做网站哪家好站内推广有哪些具体方式
  • 百度网站建设目标中国互联网协会
  • 做cpc不做网站可以吗广州线上教学
  • 域名解析到别人网站网络软文是什么意思
  • 郑州做网站公seo经典案例
  • 如何查做的网站排名优化大师电脑版官方
  • 无锡网站建设动态网站底部友情链接代码
  • 辽宁建设委员会网站首页怎样做网站卖自己的产品
  • 怎么做免费视频网站最新战争新闻事件今天
  • 怎么样可以做自己的网站b2b外链代发
  • 做技术分享网站有哪些seo查询软件
  • 做代码和网站网络营销的特点有哪些特点
  • 网站开发硬件环境中国国家培训网正规吗
  • 珠海斗门建设局官方网站互联网推广是什么