国内外网站建设/平台推广销售话术

首先先介绍下rabbitmq的一些基础概念

1、队列、生产者、消费者

      队列是RabbitMQ的内部对象，用于存储消息。P（生产者）生产消息并投递到队列中，C（消费者）可以从队列中获取消息并消费。

      

      多个消费者可以订阅同一个队列，这时队列中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

      

2、Exchange、Binding

      刚才我们看到生产者将消息投递到队列中，实际上这在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），再通过Binding将Exchange与Queue关联起来。

      

3、Exchange Type、Bingding key、routing key

      在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key。在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。

      生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。

      RabbitMQ常用的Exchange Type有三种：fanout、direct、topic。

      fanout:把所有发送到该Exchange的消息投递到所有与它绑定的队列中。

      direct:把消息投递到那些binding key与routing key完全匹配的队列中。

      topic:将消息路由到binding key与routing key模式匹配的队列中。

示例代码 git springboot-rabbitmq-exchange

4、direct模式实例

  4.1、添加pom文件

  <!-- rabbitmq依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

4.2、添加application.yml配置

spring:rabbitmq:host: 127.0.0.1port: 5672username: xxxpassword: xxxxpublisher-confirms: true   #消息发送到交换机确认机制,是否确认回调

4.3 ExchangeConfig配置

@Configuration
public class ExchangeConfig {/***   1.定义direct exchange，绑定queueTest*   2.durable="true" rabbitmq重启的时候不需要创建新的交换机*   3.direct交换器相对来说比较简单，匹配规则为：如果路由键匹配，消息就被投送到相关的队列*     fanout交换器中没有路由键的概念，他会把消息发送到所有绑定在此交换器上面的队列中。*     topic交换器你采用模糊匹配路由键的原则进行转发消息到队列中*   key: queue在该direct-exchange中的key值，当消息发送给direct-exchange中指定key为设置值时，*   消息将会转发给queue参数指定的消息队列*/@Beanpublic DirectExchange directExchange(){DirectExchange directExchange = new DirectExchange(RabbitMqConfig.EXCHANGE,true,false);return directExchange;}@Beanpublic TopicExchange topicExchange(){TopicExchange topicExchange = new TopicExchange(RabbitMqConfig.EXCHANGE_TOPIC,true,false);return topicExchange;}@Beanpublic FanoutExchange fanoutExchange (){FanoutExchange fanoutExchange = new FanoutExchange(RabbitMqConfig.EXCHANGE_FANOUT,true,false);return fanoutExchange;}}

4.4 QueueConfig配置

@Configuration
public class QueueConfig {@Beanpublic Queue firstQueue() {/**durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列auto-delete 表示消息队列没有在使用时将被自动删除 默认是falseexclusive  表示该消息队列是否只在当前connection生效,默认是false*/return new Queue("first-queue",true,false,false);}@Beanpublic Queue secondQueue() {return new Queue("second-queue",true,false,false);}@Beanpublic Queue topicQueue() {return new Queue("topic-queue",true,false,false);}@Beanpublic Queue topicQueue1() {return new Queue("topic-queue1",true,false,false);}@Beanpublic Queue fanoutQueue1() {return new Queue("fanout-queue1",true,false,false);}@Beanpublic Queue fanoutQueue() {return new Queue("fanout-queue",true,false,false);}}

4.5、RabbitMqConfig配置

@Configuration
public class RabbitMqConfig {/** 消息交换机的名字*/public static final String EXCHANGE = "exchangeTest";/** 消息交换机的名字*/public static final String EXCHANGE_TOPIC = "exchangeTopic";public static final String EXCHANGE_FANOUT = "exchangeFanout";/** 队列key1*/public static final String ROUTINGKEY1 = "queue_one_key1";/** 队列key2*/public static final String ROUTINGKEY2 = "queue_one_key2";public static final String ROUTINGKEY3 = "*.topic.*";public static final String ROUTINGKEY_TOPIC = "aaa.topic.*";@Autowiredprivate QueueConfig queueConfig;@Autowiredprivate ExchangeConfig exchangeConfig;/*** 连接工厂*/@Autowiredprivate ConnectionFactory connectionFactory;/**将消息队列1和交换机进行绑定*/@Beanpublic Binding binding_one() {return BindingBuilder.bind(queueConfig.firstQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTINGKEY1);}/*** 将消息队列2和交换机进行绑定*/@Beanpublic Binding binding_two() {return BindingBuilder.bind(queueConfig.secondQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTINGKEY2);}@Beanpublic Binding binding_topic() {return BindingBuilder.bind(queueConfig.topicQueue()).to(exchangeConfig.topicExchange()).with(RabbitMqConfig.ROUTINGKEY3);}@Beanpublic Binding binding_topic1() {return BindingBuilder.bind(queueConfig.topicQueue1()).to(exchangeConfig.topicExchange()).with(RabbitMqConfig.ROUTINGKEY_TOPIC);}@Beanpublic Binding binding_fanout() {return BindingBuilder.bind(queueConfig.fanoutQueue()).to(exchangeConfig.fanoutExchange());}@Beanpublic Binding binding_fanout_for_third() {return BindingBuilder.bind(queueConfig.fanoutQueue1()).to(exchangeConfig.fanoutExchange());}/*** queue listener  观察 监听模式* 当有消息到达时会通知监听在对应的队列上的监听对象* @return*/@Beanpublic SimpleMessageListenerContainer simpleMessageListenerContainer_one(){SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);simpleMessageListenerContainer.addQueues(queueConfig.firstQueue());simpleMessageListenerContainer.setExposeListenerChannel(true);simpleMessageListenerContainer.setMaxConcurrentConsumers(5);simpleMessageListenerContainer.setConcurrentConsumers(1);simpleMessageListenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认return simpleMessageListenerContainer;}/*** 定义rabbit template用于数据的接收和发送* @return*/@Beanpublic RabbitTemplate rabbitTemplate() {RabbitTemplate template = new RabbitTemplate(connectionFactory);/**若使用confirm-callback或return-callback，* 必须要配置publisherConfirms或publisherReturns为true* 每个rabbitTemplate只能有一个confirm-callback和return-callback*/template.setConfirmCallback(msgSendConfirmCallBack());//template.setReturnCallback(msgSendReturnCallback());/*** 使用return-callback时必须设置mandatory为true，或者在配置中设置mandatory-expression的值为true，* 可针对每次请求的消息去确定’mandatory’的boolean值，* 只能在提供’return -callback’时使用，与mandatory互斥*///  template.setMandatory(true);return template;}/*** 消息确认机制* Confirms给客户端一种轻量级的方式，能够跟踪哪些消息被broker处理，* 哪些可能因为broker宕掉或者网络失败的情况而重新发布。* 确认并且保证消息被送达，提供了两种方式：发布确认和事务。(两者不可同时使用)* 在channel为事务时，不可引入确认模式；同样channel为确认模式下，不可使用事务。* @return*/@Beanpublic MsgSendConfirmCallBack msgSendConfirmCallBack(){return new MsgSendConfirmCallBack();}
}

4.6 DirectExchange 模式

绑定关系如下

/**将消息队列1和交换机进行绑定*/@Beanpublic Binding binding_one() {return BindingBuilder.bind(queueConfig.firstQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTINGKEY1);}/*** 将消息队列2和交换机进行绑定*/@Beanpublic Binding binding_two() {return BindingBuilder.bind(queueConfig.secondQueue()).to(exchangeConfig.directExchange()).with(RabbitMqConfig.ROUTINGKEY2);}

生产者发送对应消息

    /*** DirectExchange 生产者 发送消息* @param uuid* @param message  消息*/public void send(String uuid,Object message) {CorrelationData correlationId = new CorrelationData(uuid);rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE, RabbitMqConfig.ROUTINGKEY2,message, correlationId);}

 消费者分别消费对应队列信息 

@Component
public class SecondConsumer {@RabbitListener(queues = {"second-queue"}, containerFactory = "rabbitListenerContainerFactory")public void handleMessage(String message) throws Exception {// 处理消息System.out.println("Second Consumer {} handleMessage :"+message);}}

消费结果

4.8 TopicExchange模式

由于在绑定队列时，绑定关系如下

 public static final String ROUTINGKEY3 = "*.topic.*";public static final String ROUTINGKEY_TOPIC = "aaa.topic.*";@Beanpublic Binding binding_topic() {return BindingBuilder.bind(queueConfig.topicQueue()).to(exchangeConfig.topicExchange()).with(RabbitMqConfig.ROUTINGKEY3);}@Beanpublic Binding binding_topic1() {return BindingBuilder.bind(queueConfig.topicQueue1()).to(exchangeConfig.topicExchange()).with(RabbitMqConfig.ROUTINGKEY_TOPIC);}

 那么此时生产者，发送aaa.topic.bbb的routing_key时。topicConsumer和topicConsumer1都能消费信息

/*** TopicExchange 生产者* @param uuid* @param message*/public void sendTopic(String uuid,Object message) {CorrelationData correlationId = new CorrelationData(uuid);rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_TOPIC, "aaa.topic.bbb",message, correlationId);}

方法的第一个参数是交换机名称,第二个参数是发送的key,第三个参数是内容,RabbitMQ将会根据第二个参数去寻找有没有匹配此规则的队列,如果有,则把消息给它,如果有不止一个,则把消息分发给匹配的队列(每个队列都有消息!),显然在我们的测试中,参数2匹配了两个队列,因此消息将会被发放到这两个队列中,而监听这两个队列的监听器都将收到消息!那么如果把参数2改为bbb.topic.bbb呢?显然只会匹配到一个队列,那么TopicConsumer方法对应的监听器收到消息!

 消费分别消费对应队列信息

@Component
public class TopicConsumer {@RabbitListener(queues = {"topic-queue"}, containerFactory = "rabbitListenerContainerFactory")public void handleMessage(String message) throws Exception {// 处理消息System.out.println("TopicConsumer {} handleMessage :"+message);}
}

 

@Component
public class TopicConsumer1 {@RabbitListener(queues = {"topic-queue1"}, containerFactory = "rabbitListenerContainerFactory")public void handleMessage(String message) throws Exception {// 处理消息System.out.println("TopicConsumer1 {} handleMessage :"+message);}
}

消费者消费结果

4.9 fanoutExchange 模式

绑定关系如下

  @Beanpublic Binding binding_fanout() {return BindingBuilder.bind(queueConfig.fanoutQueue()).to(exchangeConfig.fanoutExchange());}@Beanpublic Binding binding_fanout_for_third() {return BindingBuilder.bind(queueConfig.fanoutQueue1()).to(exchangeConfig.fanoutExchange());}

 消费者配置

@Component
public class FanoutConsumer {@RabbitListener(queues = {"fanout-queue"}, containerFactory = "rabbitListenerContainerFactory")public void handleMessage(String message) throws Exception {// 处理消息System.out.println("FanoutConsumer {} handleMessage :"+message);}
}

 

@Component
public class FanoutConsumer1 {@RabbitListener(queues = {"fanout-queue1"}, containerFactory = "rabbitListenerContainerFactory")public void handleMessage(String message) throws Exception {// 处理消息System.out.println("FanoutConsumer1 {} handleMessage :"+message);}}

生产者

   public void sendFanout(String uuid,Object message) {CorrelationData correlationId = new CorrelationData(uuid);//中间是设置路由规则，由于是广播模式，这个规则会被抛弃rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_FANOUT, "", message);}

Fanout Exchange形式又叫广播形式,因此我们发送到路由器的消息会使得绑定到该路由器的每一个Queue接收到消息,这个时候就算指定了Key,或者规则(即上文中convertAndSend方法的参数2),也会被忽略!

消费结果

5.0 消息确认回调

public class MsgSendConfirmCallBack implements RabbitTemplate.ConfirmCallback {@Overridepublic void confirm(CorrelationData correlationData, boolean b, String s) {System.out.println("MsgSendConfirmCallBack  , 回调id:" + correlationData);if (b) {System.out.println("消息发送成功");} else {System.out.println("消息发送失败:" + s+"\n重新发送");}}
}