新建我的项目
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;
@SpringBootApplication
public 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;
@Component
public 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. 音讯长久化 - 默认是长久音讯
*/
@SpringBootApplication
public 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;
@Component
public 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;
@Component
public 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;
@SpringBootApplication
public 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;
@Component
public 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;
@Component
public 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;
@SpringBootApplication
public 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;
@Component
public 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;
@Component
public 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;
@SpringBootApplication
public 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;
@Component
public 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;
@Component
public 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;
@SpringBootApplication
public 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;
@Component
public 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;
@Component
public 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);
}
}