什么网站找厂做袜子,Wordpress手机 菜单,网站升级每天自动更新,网站页面设计培训班一. AMQP协议
什么是AMQP协议 AMQP(Advanced Message Queuing Protocol,高级消息队列协议):它是进程之间传递异步消息的网络协议 AMQP工作过程 发布者通过发布消息#xff0c;通过交换机#xff0c;交换机根据路由规则将收到的消息分发交换机绑定的下消息队列#xff0c;最…一. AMQP协议
什么是AMQP协议 AMQP(Advanced Message Queuing Protocol,高级消息队列协议):它是进程之间传递异步消息的网络协议 AMQP工作过程 发布者通过发布消息通过交换机交换机根据路由规则将收到的消息分发交换机绑定的下消息队列最后AMQP代理将消息推送给订阅了此队列的消费者 或消费者按照需求自行获取。 二. RabbitMQ简介 RabbitMQ是通过Erlang语言基于AMQP协议编写的消息中间件它在分布式系统中可以解应用耦合、流量削峰、异步消息等问题。它有两个特性 队列排队和异步 应用解耦多个个应用程序之间可通过RabbitMQ作为媒介两个应用不再粘连实现解耦异步消息多个应用可通过RabbitMQ进行消息传递流量削峰在高并发情况下可以通过RabbitMQ的队列特性实现流量削峰应用场景 应用到队列特性的应用场景 排序算法、秒杀活动。应用到异步特性的应用场景 消息分发、异步处理、数据同步、处理耗时任务。 三.springBoot整合RabbitMQ
生产者端发送消息
pom文件 dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactIdversion2.6.3/version/dependencyyml文件
spring:application:name: producerrabbitmq:host: xxxusername: adminpassword: admin配置类,需要返回一个Queue,org.springframework.amqp.core.Queue下的Queue对象
Configuration
public class RabbitMqConfig {Beanprotected Queue queue(){return new Queue(myQueue);}
}使用RabbitMQ发送消息注入AmqpTemplate,调用convertAndSend()方法
class ProducerApplicationTests {Autowiredprivate AmqpTemplate amqpTemplate;Testvoid send() {for (int i 0; i 10; i) {amqpTemplate.convertAndSend(myQueue,这是发送的消息);System.out.println(发送成功);}}}消费端接收消息
配置同生产端不需要配置RabbitMqConfig接收消息时只需要使用注解RabbitMqConfigqueues属性绑定相应的队列即可。
Component
public class ReceiveService {RabbitListener(queues myQueue)public void test01(String msg){System.out.println(接收到消息1 msg);}RabbitListener(queues myQueue)public void test02(String msg){System.out.println(接收到消息2 msg);}RabbitListener(queues myQueue)public void test03(String msg){System.out.println(接收到消息3 msg);}
}四.交换器四种)
Direct Exchange:直连交换器 它是RabbitMQ的默认交换器给指定队列发消息绑定该消息队列的消费者一次获取消息 实战
/** 生产者发送消息发送10个消息*/
SpringBootTest
class ProducerApplicationTests {Autowiredprivate AmqpTemplate amqpTemplate;Testvoid send() {for (int i 0; i 10; i) {amqpTemplate.convertAndSend(myQueue,这是发送的消息);System.out.println(发送成功);}}}/** 接收消息*/
Component
public class ReceiveService {RabbitListener(queues myQueue)public void test01(String msg){System.out.println(接收到消息1 msg);}RabbitListener(queues myQueue)public void test02(String msg){System.out.println(接收到消息2 msg);}RabbitListener(queues myQueue)public void test03(String msg){System.out.println(接收到消息3 msg);}
}结果可以看到1、2、3依次接收消息
接收到消息1这是发送的消息
接收到消息2这是发送的消息
接收到消息3这是发送的消息
接收到消息2这是发送的消息
接收到消息3这是发送的消息
接收到消息1这是发送的消息
接收到消息3这是发送的消息
接收到消息1这是发送的消息
接收到消息2这是发送的消息
接收到消息1这是发送的消息Fanout Exchange:扇形交换器 绑定该交换器的所有队列都可以接收到消息扇形交换机将消息广播到所有与之绑定的队列。无论消息的路由键是什么扇形交换机都会将消息发送到所有绑定的队列中。这种类型的交换机常用于实现发布-订阅模式将消息广播给多个消费者。 实战
/** 绑定*/
/** Fanout Exchange*/
Bean
public Queue FanoutExchangeQueue1(){return new Queue(fanoutExchangeQueue1);}
Bean
public Queue FanoutExchangeQueue2(){return new Queue(fanoutExchangeQueue2);}
Bean
public FanoutExchange fanoutExchange(){return new FanoutExchange(amq.fanout);}
Bean
public Binding FanoutExchangeBinding1(Queue FanoutExchangeQueue1,FanoutExchange fanoutExchange){return BindingBuilder.bind(FanoutExchangeQueue1).to(fanoutExchange);}
Bean
public Binding FanoutExchangeBinding2(Queue FanoutExchangeQueue2,FanoutExchange fanoutExchange){return BindingBuilder.bind(FanoutExchangeQueue2).to(fanoutExchange);}/** 生产者发送消息*/Testvoid sendByFanoutExchange() {amqpTemplate.convertAndSend(amq.fanout,key,这是发送到的消息);System.out.println(发送成功);}/** 消费者 Direct Exchange*/RabbitListener(queues fanoutExchangeQueue1)public void test04(String msg){System.out.println(接收到消息4 msg);}RabbitListener(queues fanoutExchangeQueue2)public void test05(String msg){System.out.println(接收到消息5 msg);}结果每一个绑定到Fanout Exchange上的队列都可以接收到消息
接收到消息4这是发送到的消息
接收到消息5这是发送到的消息Topic Exchange:主题交换器 允许在路由键中设置匹配规则*‘代表一个字母两个’.之间的内容‘#’代表0或多个字符 实战 /** 绑定*/Beanpublic Queue topicExchangeQueue1(){return new Queue(topicExchangeQueue1);}Beanpublic Queue topicExchangeQueue2(){return new Queue(topicExchangeQueue2);}Beanpublic TopicExchange topicExchange(){return new TopicExchange(amq.topic);}Beanpublic Binding TopicExchangeToQueue1(Queue topicExchangeQueue1,TopicExchange topicExchange){return BindingBuilder.bind(topicExchangeQueue1).to(topicExchange).with(com.shaoby.*);}Beanpublic Binding TopicExchangeToQueue2(Queue topicExchangeQueue2,TopicExchange topicExchange){return BindingBuilder.bind(topicExchangeQueue2).to(topicExchange).with(com.shaoby.test.#);}/**生产者发送消息*//** key为com.shaoby.test*/Testvoid sendByTopicExchange() {amqpTemplate.convertAndSend(amq.topic,com.shaoby.test,这是发送到的消息);System.out.println(发送成功);}/** key为com.shaoby.test.a*/Testvoid sendByTopicExchange() {amqpTemplate.convertAndSend(amq.topic,com.shaoby.test.a.b,这是发送到的消息);System.out.println(发送成功);}/**消费者接收消息*//**Topic Exchange*/RabbitListener(queues topicExchangeQueue1)public void test06(String msg){System.out.println(接收到消息6 msg);}RabbitListener(queues topicExchangeQueue2)public void test07(String msg){System.out.println(接收到消息7 msg);}结果 路由key为com.shaoby.test都能接收到消息com.shaoby.test.a.b只有topicExchangeQueue2能接收到消息 Header Exchange:首部交换器
绑定
/** Header Exchange*/
Bean
public Queue headerExchangeQueue1(){return new Queue(headerExchangeQueue1);}Bean
public Queue headerExchangeQueue2(){return new Queue(headerExchangeQueue2);}
Bean
public HeadersExchange headersExchange(){return new HeadersExchange(amp.header);}
Bean
public Binding headExchangeToQueue1(Queue headerExchangeQueue1,HeadersExchange headersExchange){HashMapString, Object map new HashMap();map.put(type,OK);map.put(status,200);return BindingBuilder.bind(headerExchangeQueue1).to(headersExchange).whereAll(map).match();}
Bean
public Binding headExchangeToQueue2(Queue headerExchangeQueue2,HeadersExchange headersExchange){HashMapString, Object map new HashMap();map.put(type,error);map.put(status,500);return BindingBuilder.bind(headerExchangeQueue2).to(headersExchange).whereAll(map).match();}
/** 生产者发送消息*/
Testvoid sendByHeadExchange() {MapString, Object headers new HashMap();headers.put(type,OK);headers.put(status,200);String message 这是发送到的消息;MessageProperties messageProperties new MessageProperties();headers.forEach(messageProperties::setHeader);Message msg new Message(message.getBytes(), messageProperties);amqpTemplate.convertAndSend(amp.header,null, msg);System.out.println(发送成功);}RabbitListener(queues headerExchangeQueue1)public void test08(Message msg){System.out.println(接收到消息8: msg.toString());}RabbitListener(queues headerExchangeQueue2)public void test09(Message msg){System.out.println(接收到消息9: msg.toString());}结果只有匹配上header才能收到消息
接收到消息8:(Body:[Ba7b38a8(byte[24]) MessageProperties [headers{typeOK, status200}, contentTypeapplication/octet-stream, contentLength0, receivedDeliveryModePERSISTENT, priority0, redeliveredfalse, receivedExchangeamp.header, receivedRoutingKey, deliveryTag2, consumerTagamq.ctag-1WTdKW4n_rAEdJUosQD7bg, consumerQueueheaderExchangeQueue1])
