消息可靠性投递

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。

消息投递步骤：

1.生产者(channel)---->交换机------>队列中。

我们将利用这两个callback控制消息的可靠性投递

rabbitmq整个消息投递的路径为:
producer—>rabbitmq broker—>exchange—>queue—>consumer
消息从producer 到exchange则会返回一个confirmCallback .
消息从exchange–>queue投递失败则会返回一个returnCallback .

确认模式

springboot开启确认模式配置

spring:
  rabbitmq:
    host: 192.168.31.124
    #开启rabbitMQ的生产方确认模式
    publisher-confirm-type: correlated

设置RabbitTemplate的确认回调函数

@Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 发送方到交换机的可靠性。
     *    1.开启confirm模式，publisher-confirm-type: correlated
     *    2.设置rabbitTemplate的确认回调函数。如果消息到达交换机则返回true,如果消息没有到达交换机则返回false
     */
    @Test
    public void confirmTest(){

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                if(!b){//消息没有到达交换机  根据业务需求。
                    System.out.println("没有收到");
                    //业务操作
                }
            }
        });
        rabbitTemplate.convertAndSend("exchange001","","hello");
    }

退回模式

springboot配置退回机制

server:
  port: 8081

spring:
  rabbitmq:
    host: 192.168.31.124
    #开启rabbitMQ的生产方确认模式
    publisher-confirm-type: correlated
    # 开启发布者退回模式
    publisher-returns: true

(2)设置RabbitTemplate回调的函数

/**
     *  退回模式:
     *     1. 开启退回模式。
     *     2. 设置RabbitTemplate的退回回调函数。
     */
    @Test
    public void returnTest(){
            rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
                @Override
                public void returnedMessage(ReturnedMessage returnedMessage) {
                    //只要交换机到队列失败时才会触发该方法。 可以继续发送也可以取消相应的业务功能。
                    System.out.println("消息从交换机到队列失败"+returnedMessage.getReplyText());
                }
            });
        rabbitTemplate.convertAndSend("direct001","error","hello2");
    }

Consumer ACK（消息确认方式）

表示消费端收到消息后的确认方式。

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息队列中移除。
但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

springboot为消费端配置手动开启确认模式

spring:
  rabbitmq:
    host: 192.168.31.124
    listener:
      simple:
        #表示手动确认
        acknowledge-mode: manual
      # 表示自动确认模式,默认为自动确认
        # acknowledge-mode: none

 @RabbitListener(queues = "direct001")
    public void listener(Message message, Channel channel) throws Exception{
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        byte[] body = message.getBody();
        String msg=new String(body);

        System.out.println(msg);
        try {
//            int c = 10 / 0;
            System.out.println("处理业务逻辑");
            //消费端手动确认消息
            channel.basicAck(deliveryTag,true); //从队列中删除该消息。
        }catch (Exception e){
        	//long deliveryTag  表示的标识。
            // boolean multiple 是否允许多确认
            // boolean requeue: 是否让队列再次发送该消息。
            channel.basicNack(deliveryTag,true,true);
        }

    }

如何保证消息可靠性？

1. 保证消息从发送者到交换机的可靠性： 使用Confirm确认机制。
2. 保证消息从交换机到队列的可靠性; 使用return回退机制。
3. 消息在队列中的可靠性。 设置队列和消息的持久化。
4. 保证消息从队列到消费者的可靠性。 使用消费端的手动确认机制。

消费端限流

1. 必须为手动确认模式。
2. 必须配置限流的个数。
   springboot配置

spring:
  rabbitmq:
    host: 192.168.31.124
    listener:
      simple:
        #表示手动确认
        acknowledge-mode: manual
      # 表示自动确认模式
        # acknowledge-mode: none
        # 设置每次消费的个数。
        prefetch: 20

@Component
public class MyListener {

    @RabbitListener(queues = "direct001")
    public void listener(Message message, Channel channel) throws Exception{
        long deliveryTag = message.getMessageProperties().getDeliveryTag();

        byte[] body = message.getBody();
        String msg=new String(body);

        System.out.println(msg);

        try {
			//int c = 1 / 0;
            //System.out.println("处理业务逻辑");
            //消费端手动确认消息
            channel.basicAck(deliveryTag,true); //从队列中删除该消息。
        }catch (Exception e){
         	// long deliveryTag  表示的标识。
            // boolean multiple 是否允许多确认
            // boolean requeue 是否让队列再次发送该消息。
            channel.basicNack(deliveryTag,true,true);
        }

    }
}

TTL

1.设置队列过期;
2.设置消息的过期；该消息必须在队列的头部时才会被移除。

 	//为队列设置过期时间  相当于该队列里面的消息都有过期时间
    @Test
    public void sendTest(){
        rabbitTemplate.convertAndSend("exchange002","","hello hh");
    }

    //设置消息的过期时间 如果设置了队列的过期时间 也设置了消息的过期时间 谁的过期时间短 以谁为准。
    //该消息必须在头部才能从队列中移除。
    @Test
    public void testSend02(){

        for(int i=0;i<10;i++) {
            if(i==6){
                MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
                    @Override
                    public Message postProcessMessage(Message message) throws AmqpException {
                        message.getMessageProperties().setExpiration("10000");
                        return message;
                    }
                };
                //String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor
                rabbitTemplate.convertAndSend("exchange002", "", "hello hh"+i, messagePostProcessor);
            }else {
                rabbitTemplate.convertAndSend("exchange002", "", "hello hh"+i);
            }
        }
    }

springboot创建队列和交换机

@Configuration
public class RabbitConfig {

    private final String exchange_name="myexchange";
    private final String queue_name="myqueue";
    //创建交换机对象
    @Bean
    public Exchange exchange(){
        Exchange exchange= ExchangeBuilder.fanoutExchange(exchange_name).durable(true).build();
        return exchange;
    }

    //创建队列
    @Bean(value = "queue")
    public Queue queue(){
        Queue queue= QueueBuilder.durable(queue_name).withArgument("x-message-ttl",20000).build();
        return queue;
    }

    //绑定交换机和队列
    @Bean
    public Binding binding(Queue queue,Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("").noargs();
    }
}

死信队列

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

消息成为死信的三种情况：

1. 队列消息长度到达限制；
2. 消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；
3. 原队列存在消息过期设置，消息到达超时时间未被消费；

队列绑定死信交换机：

给队列设置参数： x-dead-letter-exchange 和 x-dead-letter-routing-key

springboot创建死信队列

@Configuration
public class RabbitConfig {
    private final String EXCHANGE="exchange";
    private final String DEAD_EXCHANGE="dead_exchange";
    private final String QUEUE="queue";
    private final String DEAD_QUEUE="dead_queue";


    @Bean
    public Queue queue(){
        return QueueBuilder
                .durable(QUEUE)
                .withArgument("x-message-ttl",20000)
                .withArgument("x-max-length",10)
                .withArgument("x-dead-letter-exchange",DEAD_EXCHANGE)
                .withArgument("x-dead-letter-routing-key","error")
                .build()
                ;
    }

    @Bean
    public Queue dead_queue() {
        return QueueBuilder.durable(DEAD_QUEUE).build();
    }

    @Bean
    public Exchange exchange(){
        return ExchangeBuilder.directExchange(EXCHANGE).build();
    }

    @Bean
    public Exchange dead_exchange(){
        return ExchangeBuilder.directExchange(DEAD_EXCHANGE).build();
    }


    @Bean
    public Binding binding(){
        return BindingBuilder.bind(queue()).to(exchange()).with("error").noargs();
    }

    @Bean
    public Binding dead_binding(){
        return BindingBuilder.bind(dead_queue()).to(dead_exchange()).with("error").noargs();
    }

}

延迟队列

消息幂等性保障

幂等性: 无论执行多少次，得到的结果和第一次都是相同的。 根据id=2查询. 删除 id=1.
保证消息不被重复消费。
保证消息不被重复消费。

rabbitMQ集群

一台主机启动多个rabbitMQ–伪集群

先停止rabbitMQ服务

service rabbitmq-server stop

开启第一个节点

[root@super sbin]# RABBITMQ_NODE_PORT=5673 RABBITMQ_NODENAME=rabbit1 rabbitmq-server start
RabbitMQ 3.6.5. Copyright © 2007-2016 Pivotal Software, Inc.
## ## Licensed under the MPL. See http://www.rabbitmq.com/
## ##
########## Logs: /var/log/rabbitmq/rabbit1.log
###### ## /var/log/rabbitmq/rabbit1-sasl.log
##########
Starting broker…
completed with 3 plugins.

开启第二个节点

[root@super ~]# RABBITMQ_NODE_PORT=5674 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15674}]" RABBITMQ_NODENAME=rabbit2 rabbitmq-server start
RabbitMQ 3.6.5. Copyright © 2007-2016 Pivotal Software, Inc.
## ## Licensed under the MPL. See http://www.rabbitmq.com/
## ##
########## Logs: /var/log/rabbitmq/rabbit2.log
###### ## /var/log/rabbitmq/rabbit2-sasl.log
##########
Starting broker…
completed with 3 plugins.

设置主从关系

rabbit1操作作为主节点：

[root@super ~]# rabbitmqctl -n rabbit1 stop_app
Stopping node rabbit1@super …
[root@super ~]# rabbitmqctl -n rabbit1 reset
Resetting node rabbit1@super …
[root@super ~]# rabbitmqctl -n rabbit1 start_app
Starting node rabbit1@super …

rabbit2操作为从节点：

[root@super ~]# rabbitmqctl -n rabbit2 stop_app
Stopping node rabbit2@super …
[root@super ~]# rabbitmqctl -n rabbit2 reset
Resetting node rabbit2@super …
[root@super ~]# rabbitmqctl -n rabbit2 join_cluster rabbit1@‘localhost’ ###’'内是主机名换成自己的
Clustering node rabbit2@super with rabbit1@super …
[root@super ~]# rabbitmqctl -n rabbit2 start_app
Starting node rabbit2@super …

RabbitMQ镜像集群配置

上面已经完成RabbitMQ默认集群模式，但并不保证队列的高可用性，尽管交换机、绑定这些可以复制到集群里的任何一个节点，但是队列内容不会复制。虽然该模式解决一项目组节点压力，但队列节点宕机直接导致该队列无法应用，只能等待重启，所以要想在队列节点宕机或故障也能正常应用，就要复制队列内容到集群里的每个节点，必须要创建镜像队列。
镜像队列是基于普通的集群模式的，然后再添加一些策略，所以你还是得先配置普通集群，然后才能设置镜像队列，我们就以上面的集群接着做。

HaProxy负载均衡RabbitMQ

#logging options
global
	log 127.0.0.1 local0 info
	maxconn 5120
	chroot /usr/local/haproxy
	uid 99
	gid 99
	daemon
	quiet
	nbproc 20
	pidfile /var/run/haproxy.pid

defaults
	log global
	
	mode tcp

	option tcplog
	option dontlognull
	retries 3
	option redispatch
	maxconn 2000
	contimeout 5s
   
     clitimeout 60s

     srvtimeout 15s	
#front-end IP for consumers and producters

listen rabbitmq_cluster
    # haproxy暴漏的端口号
	bind 0.0.0.0:5672
	
	mode tcp
	#balance url_param userid
	#balance url_param session_id check_post 64
	#balance hdr(User-Agent)
	#balance hdr(host)
	#balance hdr(Host) use_domain_only
	#balance rdp-cookie
	#balance leastconn
	#balance source //ip
	
	balance roundrobin
	    # haproxy代理的rabbit服务
        server node1 127.0.0.1:5673 check inter 5000 rise 2 fall 2
        server node2 127.0.0.1:5674 check inter 5000 rise 2 fall 2

listen stats
    # haproxy的图形化界面
	bind 192.168.31.124:8300
	mode http
	option httplog
	stats enable
	stats uri /rabbitmq-stats
	stats refresh 5s

开启Haproxy

/usr/soft/haproxy/sbin/haproxy -f /etc/haproxy/haproxy.cfg


//查看haproxy进程状态
ps -ef | grep haproxy

访问如下地址对mq节点进行监控
http://192.168.31.124:8300/rabbitmq-statss