其他分享
首页 > 其他分享> > RabbitMQ(二)

RabbitMQ(二)

作者:互联网

(1) MQ的概念
Message Queue :消息队列,存放消息的一个容器(先进先出)
(2)MQ的优点
解耦 、 异步提速 、 削峰填谷
(3)MQ的缺点
可用性减弱 、复杂性提供、处理一致性
MQ是需要成本的,适合的地方使用。
(4)MQ的实现
RabbitMQ(erlang AMQP协议)
ActiveMQ(java JMS接口规范)
RocketMQ(java)
Kafka(大数据)
(5)MQ的工作模式
简单模式、工作队列模式、发布订阅模式、路由模式、通配符模式
(6)编写代码
JAVA
ConnectionFactory
host / port / username/ password/ virtualHost
Connection
Channel
Channel.exchangeDeclare(exchangeName,exchangeType[fanout\direct\topic])
Channel.queueDeclare(queueName,true,false,false,null);
Channel.queueBind(queue,exchange,routingKey);
Channel.basicPublish(exchange,routingKey,null,body);
Consumer consumer = new DefaultConsumer(channel){

handleDelivery…
//业务
}
Channel.basicConsume(queueName,true,consumer)

Spring整合

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
      xmlns:context="http://www.springframework.org/schema/context"
      xmlns:rabbit="http://www.springframework.org/schema/rabbit"
      xsi:schemaLocation="http://www.springframework.org/schema/beans
      http://www.springframework.org/schema/beans/spring-beans.xsd
      http://www.springframework.org/schema/context
      https://www.springframework.org/schema/context/spring-context.xsd
      http://www.springframework.org/schema/rabbit
      http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
 
   
  <rabbit:connection-factory id="connectionFactory" host="" port="" username="" password="" virtual-host=""/>
   
  <rabbit:admin connection-factory="connectionFactory"/>
   
  <!-- 队列 -->
  <rabbit:queue id="" name=""></rabbit:queue>
​
  <!--交换机 -->
  <rabbit:fanout-exchange name="">
      <rabbit:bindings>
          <rabbit:binding queue=""></rabbit:binding>
      </rabbit:bindings>
  </rabbit:fanout-exchange>
   
  <rabbit:direct-exchange name="">
      <rabbit:bindings>
          <rabbit:binding key="" queue=""></rabbit:binding>
      </rabbit:bindings>
  </rabbit:direct-exchange>
   
  <rabbit:topic-exchange name="">
      <rabbit:bindings>
          <rabbit:binding pattern="" queue=""></rabbit:binding>
      </rabbit:bindings>
  </rabbit:topic-exchange>
   
  <rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
​
  <!--rabbitTemplate.convertAndSend(exchangeName,routingKey,body)-->
   
   
  <rabbit:listener-container connection-factory="connectionFactory">
      <rabbit:listener ref="监听器类"/>
  </rabbit:listener-container>
   
  <!--
  方式一
  <bean id="" class=""/>
  方式二
  <context:component-scan base-package="" />
  @Component
  public class 监听器类 implements MessageListener{
      onMessage(Message message){
          //业务操作
      }
  }
  -->
</beans>

SpringBoot整合RabbitMQ
(1)创建一个maven项目(jar)
(2)pom.xml添加起步依赖
spring-boot-starter-amqp
(3)编写引导类
(4)application.yml

spring:
  rabbitmq:
    host: 192.168.222.130
    port: 5672
    virtual-host: /longmarch
    username: 345078978
    password: 123456

(5)通过注解来完成交换机、队列、绑定关系的声明?
@Configuration
@Bean

QueueBuilder
ExchangeBuilder
BindingBuilder

package live.longmarch.rabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @PackageName: live.longmarch.rabbitmq.config
 * @ClassName: RabbitMQConfig
 * @Author: raven
 * @Date: 2020/1/11 20:56
 * @Blame: liunian
 * @Description:
 */
@Configuration
public class RabbitMQConfig {
    /**
     * 交换机名称
     */
    public static final String ITEM_TOPIC_EXCHANGE = "item_topic_exchange";
    /**
     * 队列名称
     */
    public static final String ITEM_QUEUE = "item_queue";

    /**
     * // 声明交换机
     * @return 返回交换机 设置交换机的名称,设置可以持久化
     */
    @Bean("itemTopicExchange")
    public Exchange topicExchange(){
        return ExchangeBuilder.topicExchange(ITEM_TOPIC_EXCHANGE).durable(true).build();
    }

    /**
     * 声明队列
     * @return 返回队列,设置队列的名称
     */
    @Bean("itemQueue")
    public Queue itemQueue(){
        return QueueBuilder.durable(ITEM_QUEUE).build();
    }

    /**
     *  绑定队列和交换机的关系
     * @param queue  队列
     * @param exchange 交换机
     * @return 返回绑定关系
     */
    @Bean
    public Binding itemQueueExchange(@Qualifier("itemQueue") Queue queue, @Qualifier("itemTopicExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("item.#").noargs();
    }

}

(6)发送消息
@Autowired
private RabbitTemplate rabbitTemplate;
rabbitTemplate.convertAndSend(exchangeName,routingKey,body);
package live.longmarch.rabbitmq;

/**
 * @PackageName: live.longmarch.rabbitmq
 * @ClassName: RabbitMQTest
 * @Author: raven
 * @Date: 2020/1/11 21:10
 * @Blame: liunian
 * @Description:
 */

import live.longmarch.rabbitmq.config.RabbitMQConfig;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
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 RabbitMQTest {

        @Autowired
        private RabbitTemplate rabbitTemplate;

    @Test
    public void test(){
        rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.insert", "商品新增,routing key 为item.insert");
        rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.update", "商品修改,routing key 为item.update");
        rabbitTemplate.convertAndSend(RabbitMQConfig.ITEM_TOPIC_EXCHANGE, "item.delete", "商品删除,routing key 为item.delete");
    }

}

(7)接收消息

@RabbitListener
public void method(Message message){
​
//message.getBody()--->byte[]

}
package live.longmarch.rabbitmq.listener;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

/**
 * @PackageName: live.longmarch.rabbitmq.listener
 * @ClassName: MyListener
 * @Author: raven
 * @Date: 2020/1/12 8:56
 * @Blame: liunian
 * @Description:
 */
@Component
public class MyListener {
    @RabbitListener(queues = "item_queue")
    public void myListener1(String message){
        System.out.println("消费者接收到的消息为:" + message);
    }
}

生产者工程:
application.yml文件配置RabbitMQ相关信息;
在生产者工程中编写配置类,用于创建交换机和队列,并进行绑定
注入RabbitTemplate对象,通过RabbitTemplate对象发送消息到交换机
消费者工程:
application.yml文件配置RabbitMQ相关信息
创建消息处理类,用于接收队列中的消息并进行处理

消息的可靠性

在使用 RabbitMQ 的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。
confirm 确认模式
return 退回模式
rabbitmq 整个消息投递的路径为:
producer —> rabbitmq broker —> exchange —> queue —> consumer
消息从 producer 到 exchange 则会返回一个 confirmCallback 。
让生产者能够知道消息是否发送到了Exchange
消息从 exchange 到 queue 投递失败则会返回一个 returnCallback 。
我们将利用这两个 callback 控制消息的可靠性投递
把交换机发送消息到队列中失败的消息结果告诉生产者
在这里插入图片描述
Confirm确认模式

  /**
     * 确认模式:
     * 步骤:
     * 1. 确认模式开启:ConnectionFactory中开启publisher-confirms="true"
     * 2. 在rabbitTemplate定义ConfirmCallBack回调函数
     */
(1)配置文件中
<rabbit:connection-factory id="connectionFactory"
                              host="192.168.200.131"
                              port="5672"
                              username="guest"
                              password="guest"
                              virtual-host="/"
                              publisher-confirms="true" />

(2)发送消息的时候,通过RabbitTemplate来设置确认回调
@Test
    public void testConfirm(){

       rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
           @Override
           public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                //ack:是否成功
               //cause:日志记录
               if (ack) {
                    //接收成功
                    System.out.println("接收成功消息" + cause);
                } else {
                    //接收失败
                    System.out.println("接收失败消息" + cause);
                    //做一些处理,让消息再次发送。
                }
           }
       });
        rabbitTemplate.convertAndSend(
                "test_exchange_confirm","test_routingKey_confirm3","message confirm...");
    }

Return回退模式

/**
 * 步骤:
 * 1. 开启回退模式:publisher-returns="true"
 * 2. 设置ReturnCallBack
 * 3. 设置Exchange处理消息的模式:
 *  1. 如果消息没有路由到Queue,则丢弃消息(默认)
 *  2. 如果消息没有路由到Queue,返回给消息发送方ReturnCallBack
 */
(1)配置文件中
<rabbit:connection-factory id="connectionFactory"
                               host="192.168.200.131"
                               port="5672"
                               username="guest"
                               password="guest"
                               virtual-host="/"
                               publisher-returns="true" />

(2)发送消息的时候,通过RabbitTemplate来设置回退回调
	@Test
    public void testReturn(){
        //设置交换机处理失败消息的模式
        rabbitTemplate.setMandatory(true);

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            /**
             * message:消息对象
             * replyCode:错误码
             * replyText:错误信息
             * exchange:交换机
             * routingKey:路由
             */
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("return 执行了...");
                System.out.println(message);
                System.out.println(replyCode);
                System.out.println(replyText);
                System.out.println(exchange);
                System.out.println(routingKey);
            }
        });
        rabbitTemplate.convertAndSend(
                "test_exchange_confirm","test_routingKey_confirm","message confirm...");
    }

对于确认模式:
设置ConnectionFactory的publisher-confirms=“true” 开启 确认模式。
使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack,如果为true,则发送成功,如果为false,则发送失败,需要处理。
对于退回模式
设置ConnectionFactory的publisher-returns=“true” 开启 退回模式。
使用rabbitTemplate.setReturnCallback设置退回函数,当消息从exchange路由到queue失败后,如果设置了rabbitTemplate.setMandatory(true)参数,则会将消息退回给producer。并执行回调函数returnedMessage。

Consumer的ACK确认模式
ack指 Acknowledge,确认。 表示消费端收到消息后的确认方式。
有三种确认方式:
• 自动确认:acknowledge=“none”
• 手动确认:acknowledge=“manual”
• 根据异常情况确认:acknowledge=“auto”,(这种方式使用麻烦,不作讲解)
其中自动确认是指,当消息一旦被Consumer接收到,则自动确认收到,并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中,很可能消息接收到,业务处理出现异常,那么该消息就会丢失。
如果设置了手动确认方式,则需要在业务处理成功后,调用channel.basicAck(),手动签收,如果出现异常,则调用channel.basicNack()方法,让其自动重新发送消息。

消费者的配置文件中
<rabbit:listener-container connection-factory="connectionFactory"
                               acknowledge="manual/none">
    </rabbit:listener-container>
编写监听器类
@Component
public class AckListener implements ChannelAwareMessageListener{

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        System.out.println(new String(message.getBody()));
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try{
            Thread.sleep(1000);
            //处理业务逻辑
            //System.out.println("处理出错误了");
            //int i = 10/0;
            System.out.println("处理业务逻辑");

            //channel.basicAck(deliveryTag,true);
        }catch (Exception e){
            channel.basicNack(deliveryTag,true,true);
        }
    }
}

在rabbit:listener-container标签中设置acknowledge属性,设置ack方式 none:自动确认,manual:手动确认
如果在消费端没有出现异常,则调用channel.basicAck(deliveryTag,false);方法确认签收消息
如果出现异常,则在catch中调用 basicNack或 basicReject,拒绝消息,让MQ重新发送消息。

消费的并发处理设置
可以通过MQ中的 listener-container 配置属性
perfetch = 1,表示消费端每次从mq拉去一条消息来消费,直到手动确认消费完毕后,才会继续拉去下一条消息。

<rabbit:listener-container connection-factory="connectionFactory"
                               acknowledge="manual" prefetch=“200”>
    </rabbit:listener-container>

队列的TTL设置
TTL 全称 Time To Live(存活时间/过期时间)。当消息到达存活时间后,还没有被消费,会被自动清除。
RabbitMQ可以对消息设置过期时间,也可以对整个队列(Queue)设置过期时间。

<rabbit:queue id="xx" name="xx" auto-declare="true" durable="true">
        <rabbit:queue-arguments>
            <entry key="x-message-ttl" value="10000" value-type="java.lang.Integer" />
        </rabbit:queue-arguments>
    </rabbit:queue>

    <rabbit:direct-exchange name="yy">
        <rabbit:bindings>
            <rabbit:binding key="routingKey" queue="xx"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>
    
    rabbitTemplate.convertAndSend("yy","routingKey",body);

直接丢弃消息不合适。

如果设置了消息的过期时间,也设置了队列的过期时间,它以时间短的为准。
队列过期后,会将队列所有消息全部移除。
消息过期后,只有消息在队列顶端,才会判断其是否过期(移除掉)
小结
设置队列过期时间使用参数:x-message-ttl,单位:ms(毫秒),会对整个队列消息统一过期。
设置消息过期时间使用参数:expiration。单位:ms(毫秒),当该消息在队列头部时(消费时),会单独判断这一消息是否过期。
如果两者都进行了设置,以时间短的为准。

死信队列、
死信队列,英文缩写:DLX 。Dead Letter Exchange(死信交换机),当消息成为Dead message后,可以被重新发送到另一个交换机,这个交换机就是DLX。

死信队列:未被正常消费的消息存储的容器
(1)超时未被处理 (2)超出队列长度 (3)消费者处理失败 channel.basicNack() /basicReject()–>false不自动进入队列

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述

延迟队列
问题: 消息的消费在指定的时间之后进行。
实现: TTL(设置消息的过期时间) +DLX(消息进入死信队列)
延迟队列:即消息进入队列后不会立即被消费,只有到达指定时间后,才会被消费。
Rabbitmq实现延迟队列:实现: TTL(设置消息的过期时间) +DLX(消息进入死信队列)

在这里插入图片描述
在这里插入图片描述
消息补偿
作用: 确保消息一定发送出去并且被正确消费
在这里插入图片描述
优化:
在这里插入图片描述
消息幂等性
幂等性:同一个消息被消费多次,最终得到的结果应该是一样。
解决方案: N个相同的消息,只会执行一个消息的内容,其他的消息,想办法把的操作排除掉。
数据库
table
id name price … version【int】
1 xx xxxx 1

N个相同的消息
第一次消息
update table set version = version +1 where id = 1 where version = 1
---->version 2
所有的其他消息
update table set version = version +1 where id = 1 where version = 1

Redis
N个相同的消息
第一个消息:
redis set(key,value);
所有的其他消息
redis get(key)—>结果
获取到了 : 过
获取不到 : 执行业务操作,将key存入Redis

在这里插入图片描述

总结:

SpringBoot整合RabbitMQ [掌握]
》创建一个maven项目(jar)
》pom.xml
parent: spring-boot-starter-parent
dependency: spring-boot-starter-amqp
dependency: spring-boot-starter-test
》application.yml
rabbitMQ连接的五要素:
host / port / username/ password / virtualHost
》引导类
生产者
@Configuration @Bean
Queue(QueueBuilder) Exchange(ExchangeBuilder) Binding(BindingBuilder)

	@Autowired
	private RabbitTemplate rabbitTemplate;
	
	rabbitTemplate.convertAndSend(exchangeName,routingKey,body);
消费者
	@RabbitListener(queues="queueName")
	public void method(Message message){
		//message消息对象  getBody()
	}

-------------------------面试(说)-------------------------------
消息投递的可靠性(生产者)
(1) producer----->Exchange
confirm
成功发送到Exchange
未成功发送到Exchange
???
(2) Exchange----->Queue
return
当交换机未正确发送消息到队列的时候
(3) RabbitMQ(broker)是否正常
集群

消息的限流(消费者)
(4) Consumer---->Queue获取到消息
Ack
自动确认
手动确认
channel.basicAck()
channel.basicNack()
(1)消息确认方式一定得是手动确认
(2)prefetch=“200”

TTL
设定消息队列Queue中【消息】的存活时间

死信队列
用来存储处理失败的消息
》超时的(TTL)
》超出队列存储容量(MaxLength)
》消费者通过channel.basicNack()/basicReject()

应用:
消息补偿
确保消息100%能够正确被消费
Producer记录所有发送的消息
Consumer记录所有正确消费的消息
Producer记录所有发送的消息-Consumer记录所有正确消费的消息=未被正确消息的消息(交给消费者重新发送)
Quartz/SpringTask
Redis(Set|sdiff)/Mysql
消息幂等性
同一个消息被消费多次,最终得到的结果一致
第一个消费消息的时候,执行具体的业务,其他的消费都直接过掉[不要再去执行之前的业务]
数据库的乐观锁
Redis的Key [单线程]
-------------------------了解----------------------------
日志监控
消息追踪
集群

树叶要走风怎么挽留 发布了60 篇原创文章 · 获赞 1 · 访问量 402 私信 关注

标签:rabbitTemplate,exchange,队列,RabbitMQ,消息,message,public
来源: https://blog.csdn.net/weixin_44993313/article/details/103944052