课堂笔记：RabbitMQ（四）Rabbit特性-消息确认-确认消费

三、RabbitMq特性

1、配置发送方消息确认（producer端）：

1、开启发送方消息确认机制！

 connectionFactory.setPublisherConfirms(true);

2、在proeucter中配置rabbitMQTemplate的时候设置消息发送的确认回调函数。

@Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//        rabbitTemplate.setConnectionFactory(connectionFactory);

        // 设置消息发送的确认回调，此时如果要进行数据库的修改等操作，可以自己写一个callback类，自己实现方法
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                // CorrelationData springboot提供的参数，不是必须的。失败后返回的一个信息，一般就可以将失败的订单ID返回
         		// ack: 该消息是否成功发送到rabbitMQ
         		// causr: 失败原因
                System.out.println("ack" + ack);
                System.out.println("caust " + cause);
                System.out.println("correlationData " + correlationData);
            }
        });


        // 设置消息转换器，将对象转换为JSON字符串
        rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        return rabbitTemplate;
    }

此时如果消息发送成功的话，就会执行消息确认回调函数confirm 。

补充：配置多个rabbitTemplate和设置对应的callback

是否有多个callback回调:如果注入的时候有多个rabbitTemplate的bean，此时bean修改为不再是默认的单例模式，然后我们在使用对应的rabbitTemplate的时候就需要对当前这个rabbitTemplate进行对应的confirm Callback设置。

1、首先我们需要修改rabbitTemplate在spring中不为单例模式

 @Bean
 @Scope(value = "prototype")
 public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {						      		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
 	…………
 }

2、在使用rabbitTemplate的时候，需要使用前置处理，来指定当前rabbitTemplate的conifrm callback。

@Component
public class RabbitmqMessageSend {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @PostConstruct  // 后置处理，此时在实例化这个类之前，spring都会去首先执行该方法。
    public void initCllback() {
        // 这里写回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                //
            }
        });
    }

这样每次发送消息后，都会打印出是否发送成功的消息：

acktrue
caust null
correlationData CorrelationData [id=订单ID]

如果发送失败：条件，消息刚刚已经发送了，但是瞬间MQ关闭了。

// 模拟条件，在发送后的一瞬间，关闭spring容器
@SpringBootApplication
public class AppProducer {
    public static void main(String[] args) {
//        SpringApplication.run(AppProducer.class);

        AnnotationConfigApplicationContext annotationConfigApplicationContext =
                new AnnotationConfigApplicationContext(RabbitmqConfig.class);
        RabbitmqMessageSend bean = annotationConfigApplicationContext.getBean(RabbitmqMessageSend.class);
        bean.sendMessage();
        annotationConfigApplicationContext.close();
    }
}

2、如何确保消息一定发送到Rabbitmq了？

我们刚刚所讲过的例子 在正常情况下 是没问题的， 但是 实际开发中 我们往往要考虑一些非正常的情况， 我们从消息的发送开始：

默认情况下，我们不知道我们的消息到底有没有发送到rabbitmq当中， 这肯定是不可取的， 假设我们是一个电商项目的话 用户下了订单 订单发送消息给库存 结果这个消息没发送到rabbitmq当中 但是订单还是下了，这时候 因为没有消息 库存不会去减少库存， 这种问题是非常严重的， 所以 接下来就讲一种解决方案： 失败回调

失败回调， 顾名思义 就是消息发送失败的时候会调用我们事先准备好的回调函数，并且把失败的消息 和失败原因等 返回过来。

具体操作：

注意 使用失败回调也需要开启发送方确认模式 开启方式在下文

更改RabbitmqTemplate:（producer端）

@Bean
 public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
  RabbitTemplate template = new RabbitTemplate(connectionFactory);
  // 设置消息确认
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                // CorrelationData 参数是失败后返回的一个信息，一般就可以将失败的订单ID返回
                // ack:该消息是否成功发送到rabbitMQ
                System.out.println("ack" + ack);
                System.out.println("caust " + cause);
                System.out.println("correlationData " + correlationData);
            }
        });
 
  
   //开启mandatory模式（开启失败回调）
   template.setMandatory(true);
   //指定失败回调接口的实现类
   template.setReturnCallback(new MyReturnCallback());
   return template;
 }

 

回调接口的实现类：

实现RabbitTemplate.ReturnCallback里面的returnedMessage方法即可 他会吧相关的参数都传给你

public class MyReturnCallback implements RabbitTemplate.ReturnCallback {
 
   @Override
   public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
     System.out.println(message);
     System.out.println(replyCode);
     System.out.println(replyText);
     System.out.println(exchange);
     System.out.println(routingKey);
   }
 }

参数解释：

1、message：分为body和messageProperties两部分，分别是：“发送的消息 hello” + 发送消息的配置（如过期时间）,

这里模拟一个失败的发送 ： 当指定的交换机不能吧消息路由到队列时（没有指定路由建或者指定的路右键没有绑定对应的队列 或者压根就没有绑定队列都会失败） 消息就会发送失败 效果（在web页面新建一个exchange,不要设定路由键，这样会发送失败）:

http://localhost:8082/order.do?exchange=errorExchange&routingKey=debug.order.B&message=helloMQ2222

(Body:'helloMessage' MessageProperties [headers={spring_returned_message_correlation=订单ID}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
312
NO_ROUTE
errorExchange
debug.order2222.B
acktrue
caust null
correlationData CorrelationData [id=订单ID]

分别打印的是错误状态码，错误原因（这里的原因是不能路由） 交换机名字 和路由建 （还有个参数是你发送出去的消息 因为太长了就没截图了.）

这时候，如果配合消息确认和失败回调，就能确保消息一定发送成功。

注意：如果发送的消息为Map集合，可以考虑将集合进行序列化。可以自己设置消息转化器。

可能有些同学想到了一个答案------事务

没错事务的确能解决这个问题， 而且 恰巧rabbitmq刚好也支持事务， 但是！ 事务非常影响rabbitmq的性能 有多严重？ 据我所查到的资料 （当然 只是我所了解的 同学们也可以自己去尝试测试结果） 开启rabbitmq事务的话 对性能的影响超过100****倍之多 也就是说 开启事务后处理一条消息的时间 不开事物能处理100条（姑且这样认为吧）， 那么 这样是非常不合理的， 因为消息中间件的性能其实非常关键的（参考双11） 如果这样子做的话 虽然能确保消息100%投递成功 但是代价太大了！

那么除了事务还有什么解决方案吗？

rabbitmq其实还提供了一种解决方案， 叫:发送方确认模式 这种方式 对性能的影响非常小 而且也能确定消息是否发送成功

而且 发送方确认模式一般也会和失败回调一起使用 这样 就能确保消息100%投递了

发送方确认开启:

其实代码在上面配置连接的时候已经放出来了 就是在连接工厂那被注释的一行代码 ：

  connectionFactory.setPublisherConfirms(true);

如果是yml配置的话：

spring:
  rabbitmq:
   publisher-confirms: true

和失败回调一样 实现一个接口:

public class MyConfirmCallback implements RabbitTemplate.ConfirmCallback{
 
   @Override
   public void confirm(CorrelationData correlationData, boolean ack, String cause) {
     System.out.println(correlationData);
     System.out.println(ack);
     System.out.println(cause);
   }
 }

 

在RabbitmqTemplate 设置一下

template.setConfirmCallback(new MyConfirmCallback());

而且我们可以在发送消息的时候附带一个CorrelationData参数 这个对象可以设置一个id，可以是你的业务id 方便进行对应的操作

CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend("directExchange", "direct.key123123", "hello",correlationData);

效果：

这里会吧我们传入的那个业务id 以及ack（是否发送成功） 以及原因 返回回来

但是 要注意的是 confirm模式的发送成功 的意思是发送到RabbitMq（Broker）成功 而不是发送到队列成功

所以才有了上面我所说的那句 要和失败回调结合使用 这样才能确认消息投递成功了

可能这里有点绕， 简单的总结一下就是 confirm机制是确认我们的消息是否投递到了 RabbitMq（Broker）上面 而mandatory是在我们的消息进入队列失败时候不会被遗弃（让我们自己进行处理）

那么上面 就是rabbitmq在发送消息时我们可以做的一些处理， 接下来我们会讲到rabbitmq在接收（消费）消息时的一些特性

2.1如何传递对象

可以使用FastJson来将对象序列化为字符串，然后在consumer端解析该JSON字符串为对象。

1、手动配置FastJSON转换器，SpringBoot默认会使用Jackson

public void sendMessage() {
        // 该参数可以选择是否传递，根据自己的业务需求。用于失败回调返回
        CorrelationData correlationData = new CorrelationData("订单ID");
        Map<String, Object> map = new HashMap<>();
        map.put("name", "123");
        map.put("age", 13);


        // 将map序列化为json字符串|  springboot默认的序列化实现是Jackson
        rabbitTemplate.convertAndSend("errorExchange", "debug.order.B", JSON.toJSONString(map), correlationData);
    }

@RabbitListener(queues = "test queue")
    public void getMessageForQueue2(Message message) throws Exception {
        // 获取到消息的body字节数据，然后将其转化为字节数组
        System.out.println("消费者2：" + new String(message.getBody(), "UTF-8"));

        // 也可以将接受的JSON字符串解析成相应的对象
//        User user = JSON.parseObject(new String(message.getBody(), "UTF-8"), User.class);
    }

2、配置自定义转换器，在创建rabbitTemplate的时候设置

(1)、设置发送时的转换器

 // 设置消息转换器，将对象转换为JSON字符串
        rabbitTemplate.setMessageConverter(new MessageConverter() {
            @Override  // 发送消息的转换方法，在producer只需要实现这个方法即可
            public Message toMessage(Object o, MessageProperties messageProperties) throws MessageConversionException {
                Message message = new Message(JSON.toJSONBytes(o), messageProperties);
                return message;
            }

            @Override // 接收消息的转换方法，在consumer可以选择实现
            public Object fromMessage(Message message) throws MessageConversionException {
                return null;
            }
        });

(2)、设置接收时的转换器

(3)、问题：可能会出现消息被spring转化成一串数字的情况：

@RabbitListener(queues = "test queue")
    public void getMessageForQueue(String message) throws Exception {
        System.out.println("消费者1：" + message);
    }


    @RabbitListener(queues = "test queue")
    public void getMessageForQueue2(Message message) throws Exception {
        // 获取到消息的body字节数据，然后将其转化为字节数组
        System.out.println("消费者2：" + new String(message.getBody(), "UTF-8"));

        // 也可以将接受的JSON字符串解析成相应的对象
//        User user = JSON.parseObject(new String(message.getBody(), "UTF-8"), User.class);
    }

消费者2：{"name":"123","age":13}
消费者1：123,34,110,97,109,101,34,58,34,49,50,51,34,44,34,97,103,101,34,58,49,51,125

此时没有自动转换的原因是在源码中没有满足转换条件，因为此时的contentType不是Text开头的，所以我们需要手动设置一下。

此时由于消息的ContentType不是text开头的，而是application/stream，所以会将字节数组直接返回。 所以我们只需要在发送消息的设置一下message的properties即可。

  rabbitTemplate.setMessageConverter(new Jackson2JsonMessageConverter());
        rabbitTemplate.setMessageConverter(new MessageConverter() {
            @Override  // 发送消息的转换方法，在producer只需要实现这个方法即可
            public Message toMessage(Object o, MessageProperties messageProperties) throws MessageConversionException {
                Message message = new Message(JSON.toJSONBytes(o), messageProperties);
                // 防止乱码，手动设置消息的contentType
                messageProperties.setContentType("text/xml");
                messageProperties.setContentEncoding("UTF-8");
                return message;
            }
        });

消费者1：{"name":"123","age":13}
消费者2：{"name":"123","age":13}
消费者1：{"name":"123","age":13}

2.2 备用交换机

1、这里的备用交换机必须在创建交换机的同时创建出来：

2、备用交换机的名称必须是一个已经存在了的交换机！

3、消费者如何确认消费？

为什么要确认消费？ 默认情况下 消费者在拿到rabbitmq的消息时 已经自动确认这条消息已经消费了， 讲白话就是rabbitmq的队列里就会删除这条消息了， 但是 我们实际开发中 难免会遇到这种情况， 比如说 拿到这条消息 发现我处理不了 比如说 参数不对， 又比如说 我当前这个系统出问题了， 暂时不能处理这个消息， 但是 这个消息已经被你消费掉了 rabbitmq的队列里也删除掉了， 你自己这边又处理不了， 那么 ，这个消息就被遗弃了。 这种情况在实际开发中是不合理的， rabbitmq提供了解决这个问题的方案， 也就是我们上面所说的confirm模式 只是我们刚刚讲的是发送方的 这次我们来讲消费方的。

首先 我们在消费者这边（再强调一遍 我这里建议大家消费者和生产者分两个项目来做，包括我自己就是这样的， 虽然一个项目也可以，我觉得分开的话容易理解一点）

设置一下消息确认为手动确认:

当然 我们要对我们的消费者监听器进行一定的配置的话， 我们需要先实例一个监听器的Container 也就是容器， 那么我们的监听器（一个消费者里面可以实例多个监听器） 可以指定这个容器 那么我们只需要对这个Container（容器） 进行配置就可以了

首先得声明一个容器并且在容器里面指定消息确认为手动确认:

@Bean
 public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(ConnectionFactory connectionFactory){
   SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory =
       new SimpleRabbitListenerContainerFactory();
   //这个connectionFactory就是我们自己配置的连接工厂直接注入进来
   simpleRabbitListenerContainerFactory.setConnectionFactory(connectionFactory);
   //这边设置消息确认方式由自动确认变为手动确认
   simpleRabbitListenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
   return simpleRabbitListenerContainerFactory;
 }

AcknowledgeMode关于这个类 就是一个简单的枚举类 我们来看看：

3个状态 不确认 手动确认 自动确认

我们刚刚配置的就是中间那个 手动确认

既然现在是手动确认了 那么我们在处理完这条消息之后 得使这条消息确认：

@Component 
 public class TestListener { 
 
    
   //containerFactory:指定我们刚刚配置的容器 
 @RabbitListener(queues = "testQueue",containerFactory = "simpleRabbitListenerContainerFactory") 
   public void getMessage(Message message, Channel channel) throws Exception{ 
     System.out.println(new String(message.getBody(),"UTF-8")); 
     System.out.println(message.getBody()); 
     //这里我们调用了一个下单方法 如果下单成功了 那么这条消息就可以确认被消费了 
     boolean f =placeAnOrder(); 
     if (f){ 
       //传入这条消息的标识， 这个标识由rabbitmq来维护 我们只需要从message中拿出来就可以 
       //第二个boolean参数指定是不是批量处理的 什么是批量处理我们待会儿会讲到 
       channel.basicAck(message.getMessageProperties().getDeliveryTag(),false); 
     }else { 
       //当然 如果这个订单处理失败了 我们也需要告诉rabbitmq 告诉他这条消息处理失败了 可以退回 也可以遗弃 要注意的是 无论这条消息成功与否 一定要通知 就算失败了 如果不通知的话 rabbitmq端会显示这条消息一直处于未确认状态，那么这条消息就会一直堆积在rabbitmq端 除非与rabbitmq断开连接 那么他就会把这条消息重新发给别人 所以 一定要记得通知！ 
       //前两个参数 和上面的意义一样， 最后一个参数 就是这条消息是返回到原队列 还是这条消息作废 就是不退回了。 
       channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true); 
       //其实 这个API也可以去告诉rabbitmq这条消息失败了 与basicNack不同之处 就是 他不能批量处理消息结果 只能处理单条消息  其实basicNack作为basicReject的扩展开发出来的 
       //channel.basicReject(message.getMessageProperties().getDeliveryTag(),true); 
     } 
   } 
 }

正常情况下的效果， 我就不演示给大家看了， 这里给大家看一个如果忘记退回消息的效果:

这里 我把消息确认的代码注释掉:

// channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

然后调用生产者发送一条消息 我们来看管理页面：

这里能看到 有一条消息在rabbitmq当中 而且状态是ready

然后我们使用消费者来消费掉他 注意 这里我们故意没有告诉rabbitmq我们消费成功了 来看看效果

这里 消费的结果打印就不截图了 还是来看管理页面：

就算我们消费端消费了下次 但是能看到 这条消息还是会在rabbitmq当中 只是他的状态为 unacked 就是未确认

这就是我们刚刚说的那种情况 无论消费成功与否 一定要通知rabbitmq 不然就会这样 一直囤积在rabbitmq当中 直到连接断开为止.

4、消息预取

扯完消息确认 我们来讲一下刚刚所说的批量处理的问题

什么情况下回遇到批量处理的问题呢？

在这里 就要先扯一下rabbitmq的消息发放机制了

rabbitmq 默认 他会最快 以轮询的机制吧队列所有的消息发送给所有客户端 （如果消息没确认的话 他会添加一个Unacked的标识上图已经看过了）

那么 这种机制会有什么问题呢， 对于Rabbitmq来讲 这样子能最快速的使自己不会囤积消息而对性能造成影响， 但是 对于我们整个系统来讲， 这种机制会带来很多问题， 比如说 我一个队列有2个人同时在消费，而且他们处理能力不同， 我打个最简单的比方 有100个订单消息需要处理（消费） 现在有消费者A 和消费者B ， 消费者A消费一条消息的速度是 10ms 消费者B 消费一条消息的速度是15ms （ 当然 这里只是打比方） 那么 rabbitmq 会默认给消费者A B 一人50条消息让他们消费 但是 消费者A 他500ms 就可以消费完所有的消息 并且处于空闲状态 而 消费者B需要750ms 才能消费完 如果从性能上来考虑的话 这100条消息消费完的时间一共是750ms（因为2个人同时在消费） 但是如果 在消费者A消费完的时候 能把这个空闲的性能用来和B一起消费剩下的信息的话， 那么这处理速度就会快非常多。

这个例子可能有点抽象， 我们通过代码来演示一下

我往Rabbitmq生产100条消息 由2个消费者来消费 其中我们让一个消费者在消费的时候休眠0.5秒（模拟处理业务的延迟） 另外一个消费者正常消费 我们来看看效果：

正常的那个消费者会一瞬间吧所有消息（50条）全部消费完（因为我们计算机处理速度非常快） 下图是加了延迟的消费者：

可能我笔记里面你看不出效果，这个你自己测试就会发现 其中一个消费者很快就处理完自己的消息了 另外一个消费者还在慢慢的处理 其实 这样严重影响了我们的性能了。

其实讲了这么多 那如何来解决这个问题呢？

我刚刚解释过了 造成这个原因的根本就是rabbitmq消息的发放机制导致的， 那么我们现在来讲一下解决方案: 消息预取

什么是消息预取？ 讲白了以前是rabbitmq一股脑吧所有消息都均发给所有的消费者（不管你受不受得了） 而现在 是在我消费者消费之前 先告诉rabbitmq 我一次能消费多少数据 等我消费完了之后告诉rabbitmq rabbitmq再给我发送数据

在代码中如何体现？

在使用消息预取前 要注意一定要设置为手动确认消息， 原因参考上面划重点的那句话。

因为我们刚刚设置过了 这里就不贴代码了， 完了之后设置一下我们预取消息的数量 一样 是在容器（Container）里面设置：

@Bean
 public SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory(ConnectionFactory connectionFactory){
   SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory =
       new SimpleRabbitListenerContainerFactory();
   simpleRabbitListenerContainerFactory.setConnectionFactory(connectionFactory);
   //手动确认消息
   simpleRabbitListenerContainerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
   //设置消息预取的数量
   simpleRabbitListenerContainerFactory.setPrefetchCount(1);
   return simpleRabbitListenerContainerFactory;
 }

那么设置完之后是什么效果呢？ 还是刚刚那个例子 还是2个消费者 因为会在消费者返回消息的确认之后 rabbitmq才会继续发送消息给客户端 而且客户端的消息累计量不会超过我们刚刚设置预取的数量， 所以我们再运行同样的例子的话 会发现 A消费者消费完99条消息了 B消费者才消费1条 （因为B消费者休眠了0.5秒才消费完{返回消息确认} 但是0.5秒之内A消费者就已经把所有消息消费完毕了 当然 如果计算机处理速度较慢这个结果可能会有差异,效果大概就是A消费者会处理大量消息）

我这里的效果就是B消费者只消费一条消息 A消费者就消费完了， 效果图就不发了 这里同学们尽量自己测试一下 或者改变一下参数看看效果。

关于这个预取的数量如何设置呢？ 我们发现 如果设置为1 能极大的利用客户端的性能（我消费完了就可以赶紧消费下一条 不会导致忙的很忙 闲的很闲） 但是， 我们每消费一条消息 就要通知一次rabbitmq 然后再取出新的消息， 这样对于rabbitmq的性能来讲 是非常不合理的 所以这个参数要根据业务情况设置

我根据我查阅到的资料然后加以测试， 这个数值的大小与性能成正比 但是有上限，与数据可靠性，以及我们刚刚所说的客户端的利用率成反比 大概如下图：

那么批量确认， 就是对于我们预取的消息，进行统一的确认。

5、死信交换机

我们来看一段代码:

channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);

我们上面解释过 这个代码是消息处理失败的确认 然后第三个参数我有解释过是消息是否返回到原队列， 那么问题来了，如果 没有返回给原队列 那么这条消息就被作废了？

rabbitmq考虑到了这个问题提供了解决方案： 死信交换机(有些人可能叫作垃圾回收器,垃圾交换机等)

死信交换机有什么用呢？ 在创建队列的时候 可以给这个队列附带一个交换机， 那么这个队列作废的消息就会被重新发到附带的交换机，然后让这个交换机重新路由这条消息

理论是这样， 代码如下：

@Bean 
   public Queue queue() { 
     Map<String,Object> map = new HashMap<>(); 
     //设置消息的过期时间 单位毫秒 
     map.put("x-message-ttl",10000); 
     //设置附带的死信交换机 
     map.put("x-dead-letter-exchange","exchange.dlx"); 
     //指定重定向的路由建 消息作废之后可以决定需不需要更改他的路由建 如果需要 就在这里指定 
     map.put("x-dead-letter-routing-key","dead.order"); 
     return new Queue("testQueue", true,false,false,map); 
   }

大概是这样的一个效果：

其实我们刚刚发现 所谓死信交换机， 只是对应的队列设置了对应的交换机是死信交换机， 对于交换机来讲， 他还是一个普通的交换机 。

下面会列出rabbitmq的常用配置：

队列配置：

参数名	配置作用
x-dead-letter-exchange	死信交换机
x-dead-letter-routing-key	死信消息重定向路由键
x-expires	队列在指定毫秒数后被删除
x-ha-policy	创建HA队列
x-ha-nodes	HA队列的分布节点
x-max-length	队列的最大消息数
x-message-ttl	毫秒为单位的消息过期时间，队列级别
x-max-priority	最大优先值为255的队列优先排序功能

消息配置：

参数名	配置作用
content-type	消息体的MIME类型，如application/json
content-encoding	消息的编码类型
message-id	消息的唯一性标识，由应用进行设置
correlation-id	一般用做关联消息的message-id，常用于消息的响应
timestamp	消息的创建时刻，整形，精确到秒
expiration	消息的过期时刻， 字符串，但是呈现格式为整型，精确到秒
delivery-mode	消息的持久化类型，1为非持久化，2为持久化，性能影响巨大
app-id	应用程序的类型和版本号
user-id	标识已登录用户，极少使用
type	消息类型名称，完全由应用决定如何使用该字段
reply-to	构建回复消息的私有响应队列
headers	键/值对表，用户自定义任意的键和值
priority	指定队列中消息的优先级

Rabbitmq linux安装&集群高可用

rabbitmq linux下安装

这里考虑到可能有同学没了解过linux 或者不太熟悉linux 所以下载地址之类的东西我这里直接贴现成的， 也就是说 只要按照我的步骤走下去基本上都没问题.

在安装(搭建集群)之前 确定两个点 **1:****防火墙关掉 2:**打开网络

关闭防火墙 systemctl stop firewalld.service 禁止开机自启 systemctl disable firewalld.service

首先 还是安装erlang

下载erlang wget http://www.rabbitmq.com/releases/erlang/erlang-18.2-1.el6.x8664.rpm 安装erlang rpm -ihv http://www.rabbitmq.com/releases/erlang/erlang-18.2-1.el6.x8664.rpm

安装完erlang之后 开始rabbitmq 还是再提醒一下 先装erlang 再装rabbitmq

装Rabbitmq之前 先装一个公钥 :

rpm --import https://dl.bintray.com/rabbitmq/Keys/rabbitmq-release-signing-key.asc

装好公钥之后 下载Rabbitmq:

wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.6/rabbitmq-server-3.6.6-1.el7.noarch.rpm

安装:

rpm -ihv rabbitmq-server-3.6.6-1.el7.noarch.rpm

安装中途可能会提示你需要一个叫socat的插件

如果提示了 就先安装socat 再装rabbitmq

安装socat:

yum install socat

至此 就装好了Rabbitmq了 可以执行以下命令启动Rabbitmq:

service rabbitmq-server start

和windows环境下一样 rabbitmq对于linux也提供了他的管理插件

安装rabbitmq管理插件:

rabbitmq-plugins enable rabbitmq_management

安装完管理插件之后 如果有装了浏览器的话 比如火狐 可以和windows一样 访问 一下 localhost:15672 可以看到一个熟悉的页面：

当然 如果你和我一样 是用虚拟机搭建的linux的话 可以用主机访问一下也是没问题的 比如说我这里虚拟机的地址为:

那么 安装 我们就讲到这 后面我们来讲集群环境的搭建 以及一些问题

rabbitmq集群搭建，配置

rabbbitmq由于是由erlang语言开发的 天生就支持分布式

rabbitmq 的集群分两种模式 一种是默认模式 一种是镜像模式

当然 所谓的镜像模式是基于默认模式加上一定的配置来的

在rabbitmq集群当中 所有的节点（一个rabbitmq服务器） 会被归为两类 一类是磁盘节点 一类是内存节点

磁盘节点会把集群的所有信息(比如交换机,队列等信息)持久化到磁盘当中，而内存节点只会将这些信息保存到内存当中 讲白了 重启一遍就没了。

为了可用性考虑 rabbitmq官方强调集群环境至少需要有一个磁盘节点， 而且为了高可用的话， 必须至少要有2个磁盘节点， 因为如果只有一个磁盘节点 而刚好这唯一的磁盘节点宕机了的话， 集群虽然还是可以运作， 但是不能对集群进行任何的修改操作（比如 队列添加，交换机添加，增加/移除 新的节点等）

具体想让rabbitmq实现集群， 我们首先需要改一下系统的hostname (因为rabbitmq集群节点名称是读取hostname的)

这里 我们模拟3个节点 :

rabbitmq1

rabbitmq2

rabbitmq3

linux修改hostname命令: hostnamectl set-hostname [name]

修改后重启一下 让rabbitmq****重新读取节点名字

然后 我们需要让每个节点通过hostname能ping通（记得关闭防火墙） 这里 我们可以修改修改一下hosts文件

关闭防火墙:

关闭防火墙 systemctl stop firewalld.service 禁止开机自启 systemctl disable firewalld.service

接下来,我们需要将各个节点的.erlang.cookie文件内容保持一致(文件路径/var/lib/rabbitmq/.erlang.cookie)

因为我是采用虚拟机的方式来模拟集群环境， 所以如果像我一样是克隆的虚拟机的话 同步.erlang.cookie文件这个操作在克隆的时候就已经完成了。

上面这些步骤完成之后 我们就可以开始来构建集群 了

我们先让rabbitmq2 加入 rabbitmq1与他构建为一个集群

执行命令( ram:使rabbitmq2成为一个内存节点 默认为:disk 磁盘节点)：

rabbitmqctl stopapp rabbitmqctl joincluster rabbit@rabbitmq1 --ram rabbitmqctl start_app

在构建的时候 我们需要先停掉rabbitmqctl服务才能构建 等构建完毕之后再启动

我们吧rabbitmq2添加完之后在rabbitmq3节点上也执行同样的代码 使他也加入进去 当然 我们也可以让rabbitmq3也作为一个磁盘节点

当执行完操作以后我们来看看效果:

随便在哪个节点打开管理页面都能看到集群环境各节点的信息;

有关集群的其他命令:

rabbitmq-server -detached 启动RabbitMQ节点 rabbitmqctl startapp 启动RabbitMQ**应用，而不是节点 rabbitmqctl stopapp 停止 rabbitmqctl status 查看状态 rabbitmqctl adduser mq 123456 rabbitmqctl setusertags mq administrator 新增账户 rabbitmq-plugins enable rabbitmqmanagement 启用RabbitMQManagement rabbitmqctl clusterstatus 集群状态 rabbitmqctl forgetclusternode rabbit@[nodeName] 节点摘除 rabbitmqctl reset application 重置

普通模式的rabbitmq集群搭建好后， 我们来说一下镜像模式

在普通模式下的rabbitmq集群 他会吧所有节点的交换机信息 和队列的元数据(队列数据分为两种 一种为队列里面的消息， 另外一种是队列本身的信息 比如队列的最大容量，队列的名称，等等配置信息， 后者称之为元数据) 进行复制 确保所有节点都有一份。

而镜像模式，则是吧所有的队列数据完全同步（当然 对性能肯定会有一定影响） 当对数据可靠性要求高时 可以使用镜像模式

实现镜像模式也非常简单 有2种方式 一种是直接在管理台控制， 另外一种是在声明队列的时候控制

声明队列的时候可以加入镜像队列参数 在上方的参数列表当中有解释 我们来讲一下管理台控制

镜像队列配置命令解释：

rabbitmqctl set_policy [-p Vhost] Name Pattern Definition [Priority]

-p Vhost**：** 可选参数，针对指定vhost下的queue进行设置 Name: policy****的名称 Pattern: queue****的匹配模式(正则表达式) Definition：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-mode ha-mode:指明镜像队列的模式，有效值为 all/exactly/nodes all：表示在集群中所有的节点上进行镜像 exactly**：表示在指定个数的节点上进行镜像，节点的个数由ha-params指定** nodes：表示在指定的节点上进行镜像，节点名称通过ha-params指定** ha-params：ha-mode模式需要用到的参数** ha-sync-mode：进行队列中消息的同步方式，有效值为automatic**和manual**

这里举个例子 如果想配置所有名字开头为 policy的队列进行镜像 镜像数量为1那么命令如下:

后者称之为元数据) 进行复制 确保所有节点都有一份。

而镜像模式，则是吧所有的队列数据完全同步（当然 对性能肯定会有一定影响） 当对数据可靠性要求高时 可以使用镜像模式

实现镜像模式也非常简单 有2种方式 一种是直接在管理台控制， 另外一种是在声明队列的时候控制

声明队列的时候可以加入镜像队列参数 在上方的参数列表当中有解释 我们来讲一下管理台控制

镜像队列配置命令解释：

rabbitmqctl set_policy [-p Vhost] Name Pattern Definition [Priority]

-p Vhost**：** 可选参数，针对指定vhost下的queue进行设置 Name: policy****的名称 Pattern: queue****的匹配模式(正则表达式) Definition：镜像定义，包括三个部分ha-mode, ha-params, ha-sync-mode ha-mode:指明镜像队列的模式，有效值为 all/exactly/nodes all：表示在集群中所有的节点上进行镜像 exactly**：表示在指定个数的节点上进行镜像，节点的个数由ha-params指定** nodes：表示在指定的节点上进行镜像，节点名称通过ha-params指定** ha-params：ha-mode模式需要用到的参数** ha-sync-mode：进行队列中消息的同步方式，有效值为automatic**和manual**

这里举个例子 如果想配置所有名字开头为 policy的队列进行镜像 镜像数量为1那么命令如下:

rabbitmqctl setpolicy hapolicy “^policy_” ‘{“ha-mode”:“exactly”,“ha-params”:1,“ha-sync-mode”:“automatic”}’

本文地址：https://blog.csdn.net/qq_43631716/article/details/107368903

如对本文有疑问， 点击进行留言回复！！