ToB企服应用市场:ToB评测及商务社交产业平台

标题: Java_RabbitMQ [打印本页]

作者: 盛世宏图 时间: 2024-8-28 07:28
标题: Java_RabbitMQ
RabbitMQ是一个高性能的异步通讯组件。（同步通讯就像两个人打视频电话，实时传输数据，还不能有其他人再加入，异步通讯像微信发消息，不具备实时性，也能有其他人加入。）
内容概述：

  初识MQ：

  同步调用：

    以余额付出服务为例：

  优势：

    时效性强，等待到效果才返回
  标题：

    也存在一些标题，比如拓展性差、性能下降、级联失败标题(雪崩)。
  异步调用：

以余额付出服务为例：

  优势：

    耦合度低，拓展性强
    异步调用，无需等待，性能好
    故障隔离，下游服务故障不影响上游服务
    缓存消息，流量削峰填谷
  标题：

    不能立即得到调用效果，时效性差
    不确定下游业务实行是否乐成
          业务安全依赖于Broker的可靠性
  技术选型：

  了解常用消息队列(MQ)：

    本文主要解说RabbitMQ。
  RabbitMQ：

  RabbitMQ是基于Erlang语言开发的开源消息通讯中心件，官网地址：
  RabbitMQ: One broker to queue them all | RabbitMQ
   介绍：

安装：

docker run \
-e RABBITMQ_DEFAULT_USER=han\
-e RABBITMQ_DEFAULT_PASS=123321 \
-v mq-plugins:/plugins \
--name mq \
--hostname mq \
-p 15672:15672 \
-p 5672:5672 \
--network my-net\
-d \
rabbitmq:3.8-management

复制代码

利用：

    打开RabbitMQ管理界面(通过ip与定义的端标语)，可以让交换机绑定队列，可以利用交换机模拟发送消息，也可以在绑定的队列中查看发送的消息。
  消息发送的注意事项：

  数据隔离：

    在RabbitMQ管理界面，点击Admin按钮，右侧选择Users，先新增一个用户：

    新增的用户是没有绑定虚拟主机的，也不能查看别的虚拟主机内的队列中的信息。
    接下来新建虚拟主机，注意切换到刚刚新建的用户登录然后创建，这样用户直接绑定登录用户。

  Java客户端：

  协议与规范：

    SpringAmqp官方地址：Spring AMQP
  快速入门：

1.创建队列：

2.引入依赖：

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

复制代码

3.设置信息：

spring:
rabbitmq:
host: ***.***.***.*** # 你的虚拟机IP
port: 5672 # 端口
virtual-host: /hmall # 虚拟主机
username: hmall # 用户名
password: 123 # 密码

复制代码

4.发送消息：

利用RabbitTemplate发送消息

@SpringBootTest
class SpringAmqpTest {
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testSimpleQueue() {
//1.队列名
String queueName = "simple.queue";
//2.消息
String message = "Hello,Spring Amqp!";
//3.发送消息
rabbitTemplate.convertAndSend(queueName, message);
}
}

复制代码

5.吸收消息：

利用注解@RabbitListener声明要监听的队列，监听消息。

@Slf4j
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "simple.queue")
public void listenSimpleQueue(String message) {
log.info("监听到simple.queue发送的消息: {}", message);
}
}

复制代码

WorkQueue：

如果我们只简朴的定义两个consumer监听一个queue，那么当publisher发送大量数据时，这两个consumer是不能吸收到发送的全部消息的，比如consumer1会吸收到1，3，5……，consumer2会吸收到2，4，6……，即使两个consumer处置惩罚消息速度不同等，它们也会各自完成各自要吸收的消息，然后快的会等待慢的，服从很低

消费者消息推送限定：

          这样设置完成之后，发送大量消息，两个consumer就会能者多劳，会一直处置惩罚消息不绝。
  Work模子的利用：

交换机：

Fanout交换机：

演示：

定义两个消费者绑定不同队列

@Slf4j
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String message) {
log.info("消费者1监听到fanout.queue1发送的消息: {}", message);
}
@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String message) {
log.info("消费者2监听到fanout.queue2发送的消息: {}", message);
}
}

复制代码

编写测试类，通过fanout交换机发消息

@Test
public void testFanoutQueue() {
//1.交换机名
String exchangeName = "hmall.fanout";
//2.消息
String message = "Hello,everyone!";
//3.发送消息
rabbitTemplate.convertAndSend(exchangeName, null, message);
}

复制代码

效果：

Direct交换机：

演示：

定义两个消费者绑定不同队列，绑定队列1的bindingkey为red和blue，队列2的bindingkey为red和yellow。

@Slf4j
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "direct.queue1")
public void listenDirectQueue1(String message) {
log.info("消费者1监听到direct.queue1发送的消息: {}", message);
}
@RabbitListener(queues = "direct.queue2")
public void listenDirectQueue2(String message) {
log.info("消费者2监听到direct.queue2发送的消息: {}", message);
}
}

复制代码

编写测试类，通过direct交换机发消息

@Test
public void testDirectQueue() {
//1.交换机名
String exchangeName = "hmall.direct";
//2.消息
String message = "Hello,blue!";
//3.发送消息
rabbitTemplate.convertAndSend(exchangeName, "blue", message);
}

复制代码

效果：

与Fanout差异：

Topic交换机：

演示：

定义两个消费者绑定不同队列，绑定队列1的bindingkey为china.#，队列2的bindingkey为#.news。

@Slf4j
@Component
public class SpringRabbitListener {
@RabbitListener(queues = "topic.queue1")
public void listenTopicQueue1(String message) {
log.info("消费者1监听到topic.queue1发送的消息: {}", message);
}
@RabbitListener(queues = "topic.queue2")
public void listenTopicQueue2(String message) {
log.info("消费者2监听到topic.queue2发送的消息: {}", message);
}
}

复制代码

编写测试类，通过direct交换机发消息

@Test
public void testTopicQueue() {
//1.交换机名
String exchangeName = "hmall.topic";
//2.消息
String message = "Hello,Chinese news!";
//3.发送消息
rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

复制代码

效果：

与Direct差异：

代码声明队列交换机：

new方式：

bulider方式：

@Configuration
public class FanoutConfiguration {
@Bean
public FanoutExchange fanoutExchange() {
return ExchangeBuilder.fanoutExchange("hmall.fanout").build();
}
@Bean
public Queue fanoutQueue() {
return QueueBuilder.durable("hmall.fanout").build();
}
@Bean
public Binding fanoutQueue1Binding(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
}
}

复制代码

注解方式：

消息转换器：

  这样通报的其他集合等复杂Object类对象信息可以转化为json范例，更简洁、易懂。
  业务实例改造：

1.引入依赖：

在消费者和发送者的pom.xml文件都引入依赖

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

复制代码

2.设置RabbitMQ信息：

在消费者和发送者中都需要设置，这里我把它设置到了nacos注册中心的共享设置中。

spring:
rabbitmq:
host: ***.***.***.***# 你的虚拟机IP
port: 5672 # 端口
virtual-host: /hmall # 虚拟主机
username: hmall # 用户名
password: 123 # 密码

复制代码

3.设置消息转换器

在消费者和发送者中都需要设置，我将它设置到了common模块下：

@Configuration
public class MqConfig {
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
}

复制代码

由于在其他模块扫描不到该设置类，所以利用Spring自动装配原理，将这个类放入spring.factories文件中：

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.hmall.common.config.MyBatisConfig,\
com.hmall.common.config.MvcConfig,\
com.hmall.common.config.MqConfig,\
com.hmall.common.config.JsonConfig

复制代码

4.编写消费者：

@Component
@RequiredArgsConstructor
public class PayStatusListener {
private final IOrderService orderService;
@RabbitListener(bindings = @QueueBinding(
value = @Queue(name = "trade.pay.success.queue", durable = "true"),
exchange = @Exchange(name = "pay.direct", type = "direct"),
key = "pay.success"
))
public void listenPaySuccess(Long orderId) {
orderService.markOrderPaySuccess(orderId);
}
}

复制代码

5.编写发送者：

利用try，catch发送消息，这样不管消息发送是否乐成，都不会对原代码造成影响。

@Override
@Transactional
public void tryPayOrderByBalance(PayOrderFormDTO payOrderFormDTO) {
// 1.查询支付单
PayOrder po = getById(payOrderFormDTO.getId());
// 2.判断状态
if (!PayStatus.WAIT_BUYER_PAY.equalsValue(po.getStatus())) {
// 订单不是未支付，状态异常
throw new BizIllegalException("交易已支付或关闭！");
}
// 3.尝试扣减余额
userClient.deductMoney(payOrderFormDTO.getPw(), po.getAmount());
// 4.修改支付单状态
boolean success = markPayOrderSuccess(payOrderFormDTO.getId(), LocalDateTime.now());
if (!success) {
throw new BizIllegalException("交易已支付或关闭！");
}
// TODO 5.修改订单状态
// tradeClient.markOrderPaySuccess(po.getBizOrderNo());
try {
rabbitTemplate.convertAndSend("pay.direct", "pay.success", po.getBizOrderNo());
} catch (Exception e) {
log.error("发送支付状态通知失败，订单id:{}", po.getBizOrderNo(), e);
}
}

复制代码

消息可靠性标题：

  发送者的可靠性：

  发送者重连：

注意：

发送者确认：

如何实现：

(利用@PostConstruct注解的方法必须是void范例，且不能有参数。这个方法会在对象创建并完成依赖注入后被自动调用。通常用于实行一些资源初始化、设置加载等操纵。)

  注意：

    由于发送者确认需要与MQ举行通讯与确认，所以会大大的影响消息发送的服从，不保举打开。如果要利用，也不要让消息重发无限的重试，要限定重发次数。
  MQ的可靠性：

  数据长期化：

  交换机长期化：

    默认是长期的
  队列长期化：

    和交换机长期化类似，而且默认也是长期的
  消息长期化：

    发送长期化的消息，不仅会生存到内存，还会写入磁盘中，即使MQ重启也不会消失，但是重启后非长期化的消息会消失。写代码利用Spring AMQP发送消息默认也是长期化的。
    由于发送非长期化消息时，消息过多会写入磁盘，导致吸收消息的速度呈现波浪形，服从很低。而发送长期化消息会对每条消息都长期化，而且吸收速度可以一直处于大概峰值的位置，服从更高。
  注意：

    保举打开它们的长期化，可以大大提高服从和可靠性。
  Lazy Queue：

    由于发送长期化的消息，不仅会生存到内存，还会写入磁盘中，耗时较长，就会导致整体的并发能力下降，为了解决这个标题，MQ引入了Lazy Queue。

如何利用：

  总结：

  RabbitMQ自身如何保证消息的可靠性：

  消费者的可靠性：

  消费者确认机制：

如何利用：

    注意如果是吸收数据后处置惩罚数据导致了业务异常，那么SpringAMQP是不会抛异常的，这种情况一般需要程序员自己编写代码让MQ返回nack或者reject。
  失败重试计谋：

失败消息处置惩罚计谋：

  业务幂等性：

  幂等：

  解决方案：

  1.唯一消息id：

2.业务判断：

  耽误消息：

  介绍：

    耽误消息：发送者发送消息时指定一个时间，消费者不会立即收到消息，而是在指定时间之后才收到消息。
    耽误任务：设置在肯定时间之后才实行的任务

死信交换机：

代码声明死信交换机：

代码设置发送消息逾期时间：

耽误消息插件：

  下载网址：

  rabbitmq/rabbitmq-delayed-message-exchange: Delayed Messaging for RabbitMQ (github.com)
  安装：

          由于我们是基于Docker安装，所以需要先查看RabbitMQ的插件目录对应的数据卷。

docker volume inspect mq-plugins

复制代码

效果如下：

[
{
"CreatedAt": "2024-06-19T09:22:59+08:00",
"Driver": "local",
"Labels": null,
"Mountpoint": "/var/lib/docker/volumes/mq-plugins/_data",
"Name": "mq-plugins",
"Options": null,
"Scope": "local"
}
]

复制代码

插件目录被挂载到了/var/lib/docker/volumes/mq-plugins/_data这个目录，我们上传插件到该目录下。
接下来实行下令，安装插件：

docker exec -it mq rabbitmq-plugins enable rabbitmq_delayed_message_exchange

复制代码

利用：

起首将交换机的delay属性设为true

发送消息时需要通过x-delay来设置逾期时间

  注意：

    由于每个耽误消息都有自己的计时器时钟，所以耽误消息过多对CPU压力会很大，我们应该尽量制止同一时间出现过多耽误消息，可以接纳镌汰耽误等待时间的方式，一般接纳10或15秒即10000或15000毫秒。
  面试大概出现标题：

  1.如何保证付出服务与交易服务之间的订单同等性？

  2.如果交易服务消息处置惩罚失败，有什么兜底方案？

  ①消息重试：一种常见的方法是在消息处置惩罚失败后举行重试。您可以设置一个重试机制，例如在消息处置惩罚失败后将消息重新放回队列，让消费者再次实验处置惩罚。您可以设置最大重试次数，制止无限重试导致死循环。
  ②死信队列（Dead Letter Exchange）：通过设置死信队列，可以将处置惩罚失败的消息路由到一个专门的队列中，以便进一步处置惩罚或分析失败的消息。当消息处置惩罚失败时，可以将消息发送到死信队列，然后根据需要举行处置惩罚。
  ③消息长期化：确保消息是长期化的，这样即使消费者在处置惩罚消息时发生故障，消息也不会丢失。消息长期化可以通过将消息标记为长期化并设置队列为长期化来实现。
  ④监控和报警：设置监控和报警体系来及时发现消息处置惩罚失败的情况。通过监控消息队列的状态、消费者的运行状况以及消息处置惩罚失败的次数，可以及时发现标题并采取步伐。
  ⑤人工干预：在极度情况下，如果消息处置惩罚失败的情况无法通过自动化本领解决，大概需要人工干预。在这种情况下，可以设置警报，关照相关人员参与处置惩罚。


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)