有心学做网站,那些网站可以做外链,正在建设中的网站,免费公司建站目录
一、SpringAMQP的介绍#xff1a;
二、利用SpringAMQP实现HelloWorld中的基础消息队列功能
1、因为publisher和consumer服务都需要amqp依赖#xff0c;因此这里把依赖直接放到父工程mq-demo中
2、编写yml文件
3、编写测试类#xff0c;并进行测试
三、在consumer…目录
一、SpringAMQP的介绍
二、利用SpringAMQP实现HelloWorld中的基础消息队列功能
1、因为publisher和consumer服务都需要amqp依赖因此这里把依赖直接放到父工程mq-demo中
2、编写yml文件
3、编写测试类并进行测试
三、在consumer中编写消费逻辑监听simple.queue
1、导入依赖刚才在父工程中已经导入了所以省略
2、编写yml文件
3、新建类实现消费逻辑
4、运行并测试
四、模拟WorkQueue实现一个队列绑定多个消费者
1、编写生产者生产50个消息
2、编写消费者一个消费者更快一个消费者更慢
3、测试
4、消费预取的修改
5、重新测试
五、发布和订阅
一利用SpringAMQP演示FanoutExchange的使用
1、新建config类声明交换机和队列
2、启动项目查看配置
3、编写消费者代码
4、编写生产者代码
5、运行代码观察输出
二交换机的作用
三声明队列、交换机、绑定关系的Bean是什么?
四DirectExchange
1、编写消费者代码
2、编写生产者代码
五Direct交换机与Fanout交换机的差异
六TopicExchange
1、编写消费者代码
2、编写生产者代码
3、运行测试
4、描述下Direct交换机与Topic交换机的差异
七测试发送Object类型信息
1、新增队列
2、发送对象
3、查看
4、优化使用jackson进行序列化
5、接收消息 一、SpringAMQP的介绍 AMQP是一种高级消息队列协议。 SpringAMQP是基于Spring Framework的AMQP扩展提供了一个抽象层使得使用AMQP进行消息传递变得更加简单。 SpringAMQP支持多种消息传递模式包括点对点、发布/订阅和请求/响应等。 SpringAMQP提供了许多高级功能例如队列管理、消息确认、事务和消息过滤等。 SpringAMQP提供了集成测试工具和基于Spring Boot的自动配置使得集成AMQP变得更加容易。 总之SpringAMQP是一个灵活、可扩展的AMQP实现它使得使用消息队列时变得更加容易和高效。
二、利用SpringAMQP实现HelloWorld中的基础消息队列功能
1、因为publisher和consumer服务都需要amqp依赖因此这里把依赖直接放到父工程mq-demo中
!-- AMQP依赖包含RabbitMQ--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency 2、编写yml文件
logging:pattern:dateformat: MM-dd HH:mm:ss:SSS
spring:rabbitmq:host: 192.168.248.152port: 5672virtual-host: /username: itcastpassword: 123456
3、编写测试类并进行测试 RunWith(SpringRunner.class)
SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSendMassage(){String queue simple.queue;String massageaaaaaaa;rabbitTemplate.convertAndSend(queue,massage);}
}
三、在consumer中编写消费逻辑监听simple.queue
1、导入依赖刚才在父工程中已经导入了所以省略
2、编写yml文件
logging:pattern:dateformat: MM-dd HH:mm:ss:SSS
spring:rabbitmq:host: 192.168.248.152port: 5672virtual-host: /username: itcastpassword: 123456
3、新建类实现消费逻辑 package cn.itcast.mq.listener;import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;Component
public class SpringRabbitListener {RabbitListener(queues simple.queue)public void listenSimpleQueue(String msg){System.out.println(消费者接收到消息msg);}
}4、运行并测试 注意:
消息一旦消费就会从队列删除RabbitMQ没有消息回溯功能
四、模拟WorkQueue实现一个队列绑定多个消费者
1、编写生产者生产50个消息
RunWith(SpringRunner.class)
SpringBootTest
public class SpringAmqpTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSendMassage(){String queue simple.queue;String massageHelloWorld;for (int i 0; i 50; i) {rabbitTemplate.convertAndSend(queue,massage);}}
}
2、编写消费者一个消费者更快一个消费者更慢
Component
public class SpringRabbitListener {RabbitListener(queues simple.queue)public void listenSimpleQueue(String msg){System.out.println(消费者0接收到消息msg LocalTime.now());try {Thread.sleep(20);} catch (InterruptedException e) {throw new RuntimeException(e);}}RabbitListener(queues simple.queue)public void listenSimpleQueue1(String msg){System.err.println(消费者1接收到消息__________-msg LocalTime.now());try {Thread.sleep(200);} catch (InterruptedException e) {throw new RuntimeException(e);}}
}
3、测试
我们发现虽然消费者0更快但是它并没有承担更多的工作量
这是因为消费预取机制会让消费者事先分配好要处理的消息而不是按能力分配 4、消费预取的修改
可以在yml文件中修改 listener:simple:prefetch: 1 #表示预取上限为1
5、重新测试 五、发布和订阅 一利用SpringAMQP演示FanoutExchange的使用
1、新建config类声明交换机和队列
Configuration
public class FanoutConfig {///1Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange(itcast.fanout);}Beanpublic Queue fanoutQueue1(){return new Queue(fanout.queue1);}Beanpublic Binding bindingQueue1(FanoutExchange exchange,Queue fanoutQueue1){return BindingBuilder.bind(fanoutQueue1).to(exchange);}///2Beanpublic Queue fanoutQueue2(){return new Queue(fanout.queue2);}Beanpublic Binding bindingQueue2(FanoutExchange exchange,Queue fanoutQueue2){return BindingBuilder.bind(fanoutQueue2).to(exchange);}
}
2、启动项目查看配置 绑定成功 3、编写消费者代码 RabbitListener(queues fanout.queue1)public void listenFanoutQueue1(String msg){System.err.println(消费者1接收到消息__________-msg LocalTime.now());}RabbitListener(queues fanout.queue2)public void listenFanoutQueue2(String msg){System.err.println(消费者2接收到消息__________-msg LocalTime.now());}
4、编写生产者代码 Testpublic void sendFanoutMassage(){String exchangeName itcast.fanout;String message Hello Every One;rabbitTemplate.convertAndSend(exchangeName,,message);}
5、运行代码观察输出
发现两个消费者都接收到了消息 二交换机的作用
接收publisher发送的消息将消息按照规则路由到与之绑定的队列不能缓存消息路由失败消息丢失FanoutExchange的会将消息路由到每个绑定的队列
三声明队列、交换机、绑定关系的Bean是什么?
QueueFanoutExchangeBinding
四DirectExchange 实现
1、编写消费者代码 RabbitListener(bindings QueueBinding(value Queue(name direct.queue2),exchange Exchange(name itcast.direct,type ExchangeTypes.DIRECT),key {red,blue}))public void listenDirectQueue1(String msg){System.err.println(消费者1接收到消息__________-msg LocalTime.now());}RabbitListener(bindings QueueBinding(value Queue(name direct.queue2),exchange Exchange(name itcast.direct,type ExchangeTypes.DIRECT),key {red,yellow}))public void listenDirectQueue2(String msg){System.out.println(消费者2接收到消息__________-msg LocalTime.now());} 2、编写生产者代码 Testpublic void sendDirectMassage(){String exchangeName itcast.direct;String message Hello Every One1111;rabbitTemplate.convertAndSend(exchangeName,blue,message);} Testpublic void sendDirectMassage(){String exchangeName itcast.direct;String message Hello Every One1111;rabbitTemplate.convertAndSend(exchangeName,red,message);} 五Direct交换机与Fanout交换机的差异
Fanout交换机将消息路由给每一个与之绑定的队列Direct交换机根据RoutingKey判断路由给哪个队列如果多个队列具有相同的RoutingKey则与Fanout功能类似
基于RabbitListener注解声明队列和交换机有哪些常见注解
QueueExchange
六TopicExchange 利用SpringAMQP演示TopicExchange的使用
1、编写消费者代码 RabbitListener(bindings QueueBinding(value Queue(name topic.queue1),exchange Exchange(name itcast.topic,type ExchangeTypes.TOPIC),key china.#))public void listenTopicQueue1(String msg){System.out.println(消费者1接收到消息aaaaaa__-msg LocalTime.now());}RabbitListener(bindings QueueBinding(value Queue(name topic.queue2),exchange Exchange(name itcast.topic,type ExchangeTypes.TOPIC),key #.news))public void listenTopicQueue2(String msg){System.err.println(消费者2接收到消息a__-msg LocalTime.now());}
2、编写生产者代码 Testpublic void sendTopicMassage(){String exchangeName itcast.topic;String message Hello Every One12222;rabbitTemplate.convertAndSend(exchangeName,china.news,message);}
3、运行测试 4、描述下Direct交换机与Topic交换机的差异 七测试发送Object类型信息
1、新增队列 Beanpublic Queue objectQueue(){return new Queue(object.queue);}
2、发送对象 Testpublic void sendObjectMassage(){MapString ,Object message new HashMap();message.put(name,11);message.put(age,22);rabbitTemplate.convertAndSend(object.queue,message);}
3、查看 对象被序列化了这种方式性能差不安全容易被注入 4、优化使用jackson进行序列化
引入依赖 dependencygroupIdcom.fasterxml.jackson.core/groupIdartifactIdjackson-databind/artifactId/dependency
添加配置Bean Beanpublic Jackson2JsonMessageConverter messageConverter(){return new Jackson2JsonMessageConverter();} 5、接收消息
编写配置Bean Beanpublic Jackson2JsonMessageConverter messageConverter(){return new Jackson2JsonMessageConverter();}
编写消费者代码
RabbitListener(queues object.queue)
public void listenObjectQueue1(MapString,Object msg){System.err.println(消费者接收到消息___da_______-msg LocalTime.now());
}
验证 注意
