新建我的项目
pom.xml
<?xml version="1.0" encoding="UTF-8"?><project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.2.4.RELEASE</version> <relativePath/> <!-- lookup parent from repository --> </parent> <groupId>cn.tedu</groupId> <artifactId>rabbitmq-springboot</artifactId> <version>0.0.1-SNAPSHOT</version> <name>rabbitmq-springboot</name> <description>Demo project for Spring Boot</description> <properties> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> <exclusions> <exclusion> <groupId>org.junit.vintage</groupId> <artifactId>junit-vintage-engine</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit-test</artifactId> <scope>test</scope> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build></project>application.yml
spring: rabbitmq: host: 192.168.64.140 username: admin password: admin主程序
删除主动创立的主程序
咱们为每种模式创立一个包,在每个包中创立各自的主程序,独自测试.
简略模式
主程序
Spring提供的Queue类,是队列的封装对象,它封装了队列的参数信息.
RabbitMQ的主动配置类,会发现这些Queue实例,并在RabbitMQ服务器中定义这些队列.
package cn.tedu.rabbitmqspringboot.m1;import org.springframework.amqp.core.Queue;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;import javax.annotation.PostConstruct;@SpringBootApplicationpublic class Main { @Autowired private Producer producer; public static void main(String[] args) { SpringApplication.run(Main.class, args); } @Bean public Queue helloworldQueue(){ /** * 可用以下模式： * new Queue("helloworld") - 默认属性：长久（true），非排他（false），非主动删除（false） * new Queue("helloworld",false,false,false,null) */ return new Queue("helloworld",false);//返回一个非长久队列 } /** *@PostConstruct 办法会被主动执行，spring扫描创立了所有对象，并实现所有注入操作后会执行 */ @PostConstruct public void test(){ producer.send(); System.out.println("音讯曾经发送"); }}生产者
AmqpTemplate是rabbitmq客户端API的一个封装工具,提供了简便的办法来执行音讯操作.
AmqpTemplate由主动配置类主动创立
package cn.tedu.rabbitmqspringboot.m1;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Componentpublic class Producer { // 在RabbtiAutoConfiguration 主动配置类中创立的工具对象 @Autowired private AmqpTemplate amqpTemplate; public void send(){ amqpTemplate.convertAndSend("helloworld", "Hello world!"); }}消费者
通过@RabbitListener从指定的队列接管音讯
应用@RebbitHandler注解的办法来解决音讯
package cn.tedu.rabbitmqspringboot.m1;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Component//@RabbitListener(queues = "helloworld")//这样应用须要用@RabbitHandler配合应用public class Consumer { //@RabbitHandler//这样只能接管一个队列的音讯，简略模式 @RabbitListener(queues = "helloworld")//这种用法能够写多个办法接管多个队列，工厂模式 public void receive(String msg){ System.out.println("收到： " + msg); }}工作模式
主程序
在主程序中创立名为task_queue的长久队列
package cn.tedu.rabbitmqspringboot.m2;import org.springframework.amqp.core.Queue;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;import javax.annotation.PostConstruct;/** * 正当散发 * 1.手动ack - springboot整合后默认就是手动ack模式 * 消费者办法执行胜利后，springboot会帮忙发送回执 * 2.qos=1 - yml中配置prefetch * 长久化 * 1.队列长久化 * 2.音讯长久化 - 默认是长久音讯 */@SpringBootApplicationpublic class Main { @Autowired private Producer producer; public static void main(String[] args) { SpringApplication.run(Main.class, args); } @Bean public Queue taskQueue(){ /** * 可用以下模式： * new Queue("helloworld") - 默认属性：长久（true），非排他（false），非主动删除（false） * new Queue("helloworld",false,false,false,null) */ return new Queue("task_queue",true);//返回一个长久队列 } /** *@PostConstruct 办法会被主动执行，spring扫描创立了所有对象，并实现所有注入操作后会执行 */ @PostConstruct public void test(){ producer.send(); }}生产者
package cn.tedu.rabbitmqspringboot.m2;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.Scanner;@Componentpublic class Producer { // 在RabbtiAutoConfiguration 主动配置类中创立的工具对象 @Autowired private AmqpTemplate amqpTemplate; public void send(){// new Thread(new Runnable() {// @Override// public void run() {//// }// }).start(); // lambda.匿名外部类的简写 new Thread(() ->{ while (true){//用独自的线程执行，不要影响主线程 System.out.println("输出音讯: "); String msg = new Scanner(System.in).nextLine(); amqpTemplate.convertAndSend("task_queue",msg); } }).start(); }}spring boot封装的 rabbitmq api 中, 发送的音讯默认是长久化音讯.
如果心愿发送非长久化音讯, 须要在发送音讯时做以下设置:
- 应用 MessagePostProcessor 前置处理器参数
- 从音讯中获取音讯的属性对象
- 在属性中把 DeliveryMode 设置为非长久化
//如果须要设置音讯为非长久化,能够获得音讯的属性对象,批改它的deliveryMode属性 t.convertAndSend("task_queue", (Object) s, new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { MessageProperties props = message.getMessageProperties(); props.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); return message; } });消费者
package cn.tedu.rabbitmqspringboot.m2;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class Consumer { @RabbitListener(queues = "task_queue") public void receive1(String msg){ System.out.println("消费者1-收到： " + msg); } @RabbitListener(queues = "task_queue") public void receive2(String msg){ System.out.println("消费者2-收到： " + msg); }}ack模式
在 spring boot 中提供了三种确认模式:
- NONE - 应用rabbitmq的主动确认
- AUTO - 应用rabbitmq的手动确认, springboot会主动发送确认回执 (默认)
- MANUAL - 应用rabbitmq的手动确认, 且必须手动执行确认操作
默认的 AUTO 模式中, 解决音讯的办法抛出异样, 则示意音讯没有被正确处理, 该音讯会被从新发送.
设置 ack 模式
spring: rabbitmq: listener: simple: # acknowledgeMode: NONE # rabbitmq的主动确认 acknowledgeMode: AUTO # rabbitmq的手动确认, springboot会主动发送确认回执 (默认) # acknowledgeMode: MANUAL # rabbitmq的手动确认, springboot不发送回执, 必须本人编码手动执行确认操作
如果设置为 MANUAL 模式,必须手动执行确认操作
@RabbitListener(queues="task_queue") public void receive1(String s, Channel c, @Header(name=AmqpHeaders.DELIVERY_TAG) long tag) throws Exception { System.out.println("receiver1 - 收到: "+s); // 手动发送确认回执 c.basicAck(tag, false); }抓取数量
工作模式中, 为了正当地散发数据, 须要将 qos 设置成 1, 每次只接管一条音讯, 解决实现后才接管下一条音讯.
spring boot 中是通过 prefetch 属性进行设置, 改属性的默认值是 250.
spring: rabbitmq: listener: simple: prefetch: 1 # qos=1, 默认250公布和订阅模式
主程序
创立 FanoutExcnahge 实例, 封装 fanout 类型交换机定义信息.
spring boot 的主动配置类会主动发现交换机实例, 并在 RabbitMQ 服务器中定义该交换机.
package cn.tedu.rabbitmqspringboot.m3;import org.springframework.amqp.core.FanoutExchange;import org.springframework.amqp.core.Queue;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;import javax.annotation.PostConstruct;@SpringBootApplicationpublic class Main { @Autowired private Producer producer; public static void main(String[] args) { SpringApplication.run(Main.class, args); } @Bean public FanoutExchange logsExchange(){ return new FanoutExchange("logs",false,false);//非长久，不主动删除 } @PostConstruct public void test(){ producer.send(); }}生产者
生产者向指定的交换机 logs 发送数据.
不须要指定队列名或路由键, 即便指定也有效, 因为 fanout 替换机会向所有绑定的队列发送数据, 而不是有抉择的发送.
package cn.tedu.rabbitmqspringboot.m3;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.Scanner;@Componentpublic class Producer { @Autowired private AmqpTemplate amqpTemplate; public void send(){ new Thread(() ->{ while (true){ System.out.println("输出音讯: "); String msg = new Scanner(System.in).nextLine(); amqpTemplate.convertAndSend("logs","",msg);//向交换机发送音讯 } }).start(); }}消费者
消费者须要执行以下操作:
- 定义随机队列(随机命名,非长久,排他,主动删除)
- 定义交换机(能够省略, 已在主程序中定义)
- 将队列绑定到交换机
spring boot 通过注解实现以上操作:
@RabbitListener(bindings = @QueueBinding( //这里进行绑定设置 value = @Queue, //这里定义随机队列,默认属性: 随机命名,非长久,排他,主动删除 exchange = @Exchange(name = "logs", declare = "false") //指定 logs 交换机,因为主程序中曾经定义,这里不进行定义))package cn.tedu.rabbitmqspringboot.m3;import org.springframework.amqp.rabbit.annotation.Exchange;import org.springframework.amqp.rabbit.annotation.Queue;import org.springframework.amqp.rabbit.annotation.QueueBinding;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class Consumer { //1.创立随机队列 2，指定交换机logs 3.绑定 @RabbitListener(bindings = @QueueBinding( value = @Queue(),//队列，随即名，非长久，独占，主动删除队列 exchange = @Exchange(name = "logs",declare = "false")//交换机,declare示意不定义交换机，只是应用 )) public void receive1(String msg){ System.out.println("消费者1-收到： " + msg); } //1.创立随机队列 2，指定交换机logs 3.绑定 @RabbitListener(bindings = @QueueBinding( value = @Queue(),//队列，随即名，非长久，独占，主动删除队列 exchange = @Exchange(name = "logs",declare = "false")//交换机,declare（false）示意不定义交换机，只是应用 )) public void receive2(String msg){ System.out.println("消费者2-收到： " + msg); }}路由模式
与公布和订阅模式代码相似, 只是做以下三点调整:
- 应用
direct交换机 - 队列和交换机绑定时, 设置绑定键
- 发送音讯时, 指定路由键
主程序
主程序中应用 DirectExcnahge 对象封装交换机信息, spring boot 主动配置类会主动发现这个对象, 并在 RabbitMQ 服务器上定义这个交换机.
package cn.tedu.rabbitmqspringboot.m4;import org.springframework.amqp.core.DirectExchange;import org.springframework.amqp.core.FanoutExchange;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;import javax.annotation.PostConstruct;@SpringBootApplicationpublic class Main { @Autowired private Producer producer; public static void main(String[] args) { SpringApplication.run(Main.class, args); } @Bean public DirectExchange directExchange(){ return new DirectExchange("direct_logs",false,false);//非长久，不主动删除 } @PostConstruct public void test(){ producer.send(); }}生产者
生产者向指定的交换机发送音讯, 并指定路由键.
package cn.tedu.rabbitmqspringboot.m4;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.Scanner;@Componentpublic class Producer { @Autowired private AmqpTemplate amqpTemplate; public void send(){ new Thread(() ->{ while (true){ System.out.println("输出音讯: "); String msg = new Scanner(System.in).nextLine(); System.out.println("输出路由键: "); String key = new Scanner(System.in).nextLine(); amqpTemplate.convertAndSend("direct_logs",key,msg); } }).start(); }}消费者
消费者通过注解来定义随机队列, 绑定到交换机, 并指定绑定键:
@RabbitListener(bindings = @QueueBinding( // 这里做绑定设置 value = @Queue, // 定义队列, 随机命名,非长久,排他,主动删除 exchange = @Exchange(name = "direct_logs", declare = "false"), // 指定绑定的交换机,主程序中曾经定义过队列,这里不进行定义 key = {"error","info","warning"} // 设置绑定键))package cn.tedu.rabbitmqspringboot.m4;import org.springframework.amqp.rabbit.annotation.Exchange;import org.springframework.amqp.rabbit.annotation.Queue;import org.springframework.amqp.rabbit.annotation.QueueBinding;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class Consumer { //1.创立随机队列 2，指定交换机logs 3.绑定 @RabbitListener(bindings = @QueueBinding( value = @Queue(),//队列，随即名，非长久，独占，主动删除队列 exchange = @Exchange(name = "direct_logs",declare = "false"),//交换机,declare示意不定义交换机，只是应用 key = {"error"} //设置绑定键 )) public void receive1(String msg){ System.out.println("消费者1-收到： " + msg); } //1.创立随机队列 2，指定交换机logs 3.绑定 @RabbitListener(bindings = @QueueBinding( value = @Queue(),//队列，随即名，非长久，独占，主动删除队列 exchange = @Exchange(name = "direct_logs",declare = "false"),//交换机,declare（false）示意不定义交换机，只是应用 key = {"error","info","warning"} //设置绑定键 )) public void receive2(String msg){ System.out.println("消费者2-收到： " + msg); }}主题模式
主题模式不过是具备非凡规定的路由模式, 代码与路由模式基本相同, 只做如下调整:
- 应用
topic交换机 - 应用非凡的绑定键和路由键规定
主程序
package cn.tedu.rabbitmqspringboot.m5;import org.springframework.amqp.core.DirectExchange;import org.springframework.amqp.core.TopicExchange;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;import javax.annotation.PostConstruct;@SpringBootApplicationpublic class Main { @Autowired private Producer producer; public static void main(String[] args) { SpringApplication.run(Main.class, args); } @Bean public TopicExchange directExchange(){ return new TopicExchange("topic_logs",false,false);//非长久，不主动删除 } @PostConstruct public void test(){ producer.send(); }}生产者
package cn.tedu.rabbitmqspringboot.m5;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;import java.util.Scanner;@Componentpublic class Producer { @Autowired private AmqpTemplate amqpTemplate; public void send(){ new Thread(() ->{ while (true){ System.out.println("输出音讯: "); String msg = new Scanner(System.in).nextLine(); System.out.println("输出路由键: "); String key = new Scanner(System.in).nextLine(); amqpTemplate.convertAndSend("topic_logs",key,msg); } }).start(); }}消费者
package cn.tedu.rabbitmqspringboot.m5;import org.springframework.amqp.rabbit.annotation.Exchange;import org.springframework.amqp.rabbit.annotation.Queue;import org.springframework.amqp.rabbit.annotation.QueueBinding;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class Consumer { //1.创立随机队列 2，指定交换机logs 3.绑定 @RabbitListener(bindings = @QueueBinding( value = @Queue(),//队列，随即名，非长久，独占，主动删除队列 exchange = @Exchange(name = "topic_logs",declare = "false"),//交换机,declare示意不定义交换机，只是应用 key = {"*.orange.*"} //设置绑定键 )) public void receive1(String msg){ System.out.println("消费者1-收到： " + msg); } //1.创立随机队列 2，指定交换机logs 3.绑定 @RabbitListener(bindings = @QueueBinding( value = @Queue(),//队列，随即名，非长久，独占，主动删除队列 exchange = @Exchange(name = "topic_logs",declare = "false"),//交换机,declare（false）示意不定义交换机，只是应用 key = {"*.*.rabbit","lazy.#"} //设置绑定键 )) public void receive2(String msg){ System.out.println("消费者2-收到： " + msg); }}RPC异步调用
主程序
主程序中定义两个队列
- 发送调用信息的队列:
rpc_queue - 返回后果的队列: 随机命名
package cn.tedu.m6;import java.util.UUID;import org.springframework.amqp.core.Queue;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;import org.springframework.context.annotation.Bean;@SpringBootApplicationpublic class Main { public static void main(String[] args) { SpringApplication.run(Main.class, args); } @Bean public Queue sendQueue() { return new Queue("rpc_queue",false); } @Bean public Queue rndQueue() { return new Queue(UUID.randomUUID().toString(), false); }}服务端
从rpc_queue接管调用数据, 执行运算求斐波那契数,并返回计算结果.@Rabbitlistener注解对于具备返回值的办法:
- 会主动获取
replyTo属性 - 主动获取
correlationId属性 - 向
replyTo属性指定的队列发送计算结果, 并携带correlationId属性
package cn.tedu.m6;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class RpcServer { @RabbitListener(queues = "rpc_queue") public long getFbnq(int n) { return f(n); } private long f(int n) { if (n==1 || n==2) { return 1; } return f(n-1) + f(n-2); }}客户端
应用 SPEL 表达式获取随机队列名: "#{rndQueue.name}"
发送调用数据时, 携带随机队列名和correlationId
从随机队列接管调用后果, 并获取correlationId
package cn.tedu.m6;import java.util.UUID;import org.springframework.amqp.AmqpException;import org.springframework.amqp.core.AmqpTemplate;import org.springframework.amqp.core.Message;import org.springframework.amqp.core.MessagePostProcessor;import org.springframework.amqp.core.MessageProperties;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.amqp.support.AmqpHeaders;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.beans.factory.annotation.Value;import org.springframework.messaging.handler.annotation.Header;import org.springframework.stereotype.Component;@Componentpublic class RpcClient { @Autowired AmqpTemplate t; @Value("#{rndQueue.name}") String rndQueue; public void send(int n) { // 发送调用信息时, 通过前置音讯处理器, 对音讯属性进行设置, 增加返回队列名和关联id t.convertAndSend("rpc_queue", (Object)n, new MessagePostProcessor() { @Override public Message postProcessMessage(Message message) throws AmqpException { MessageProperties p = message.getMessageProperties(); p.setReplyTo(rndQueue); p.setCorrelationId(UUID.randomUUID().toString()); return message; } }); } //从随机队列接管计算结果 @RabbitListener(queues = "#{rndQueue.name}") public void receive(long r, @Header(name=AmqpHeaders.CORRELATION_ID) String correlationId) { System.out.println("nn"+correlationId+" - 收到: "+r); } }