目录
一:交换机
1:Direct交换机
1.1生产者端代码:
1.2:消费者端代码:
2:Topic主题交换机
2.1:生产者代码:
2.2:消费者代码:
二:核心特性
2.1:消息过期机制
2.1.1:给队列中的全部消息指定过期时间
2.1.2:给某条消息指定过期时间
2.2:死信队列
一:交换机
1:Direct交换机
绑定:让交换机和队列进行关联，可以指定让交换机把什么样的消息发送给队列。
rountingkey:路由键，控制消息要发送哪个队列。
特点:根据路由键指定要转发到指定的队列
场景:特定的消息指定给特定的队列
1.1生产者端代码:
我们规定，通过控制台输入消息和路由，来指定谁完成该任务。
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.Scanner;public class DirectProducer {private static final String EXCHANGE_NAME = "2";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {//创建交换机的名称channel.exchangeDeclare(EXCHANGE_NAME, "direct");Scanner scanner=new Scanner(System.in);while(scanner.hasNext()){String userInput=scanner.nextLine();String[] s = userInput.split(" ");if(s.length<1){continue;}//指定路由keyString message=s[0];String routingKey=s[1];//发布消息/*第一个参数:发布到哪个交换机第二个参数:路由键*/channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes("UTF-8"));System.out.println("[x] Sent"+message+"with rounting"+routingKey+" ");}}}//..}1.2:消费者端代码:
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.Scanner;public class DirectProducer {private static final String EXCHANGE_NAME = "2";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {//创建交换机的名称channel.exchangeDeclare(EXCHANGE_NAME, "direct");Scanner scanner=new Scanner(System.in);while(scanner.hasNext()){String userInput=scanner.nextLine();String[] s = userInput.split(" ");if(s.length<1){continue;}//指定路由keyString message=s[0];String routingKey=s[1];//发布消息/*第一个参数:发布到哪个交换机第二个参数:路由键*/channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes("UTF-8"));System.out.println("[x] Sent"+message+"with rounting"+routingKey+" ");}}}//..}运行结果:
2:Topic主题交换机
特点:消息会根据一个模糊的路由键转发到指定的队列中。
场景:特定的一类消息只交给特定的一类系统(程序来处理)。
绑定关系:模糊匹配消息队列 *:匹配一个单词 #:匹配0个或多个单词
2.1:生产者代码:
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.util.Scanner;public class TopicProducer {private static final String EXCHANGE_NAME = "3";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.exchangeDeclare(EXCHANGE_NAME, "topic");Scanner scanner=new Scanner(System.in);while(scanner.hasNext()){String userInput=scanner.nextLine();String[] s = userInput.split(" ");if(s.length<1){continue;}//指定路由keyString message=s[0];String routingKey=s[1];//发布消息/*第一个参数:发布到哪个交换机第二个参数:路由键*/channel.basicPublish(EXCHANGE_NAME,routingKey,null,message.getBytes("UTF-8"));System.out.println("[x] Sent"+message+"with rounting"+routingKey+" ");}}}
}
2.2:消费者代码:
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;public class TopicConsumer {private static final String EXCHANGE_NAME = "3";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "topic");//创建消息队列String queueName="fronted_queue";channel.queueDeclare(queueName,true,false,false,null);channel.queueBind(queueName,EXCHANGE_NAME,"#.前端.#");String queueName2="backed-_queue";channel.queueDeclare(queueName2,true,false,false,null);channel.queueBind(queueName2,EXCHANGE_NAME,"#.后端.#");String queueName3="product_queue";channel.queueDeclare(queueName3,true,false,false,null);channel.queueBind(queueName3,EXCHANGE_NAME,"#.产品.#");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [前端] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [后端] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback deliverCallback3 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [产品] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};channel.basicConsume(queueName, true, deliverCallback1, consumerTag -> { });channel.basicConsume(queueName2, true, deliverCallback2, consumerTag -> { });channel.basicConsume(queueName3, true, deliverCallback3, consumerTag -> { });}}
运行结果:
二:核心特性
2.1:消息过期机制
特点:给每条消息指定一个有效期，一段时间内未被消费，就过期了。
2.1.1:给队列中的全部消息指定过期时间
在消费者中对于队列的全部消息指定过期时间，如果在过期时间内，还没有消费者取消息，消息才会过期，如果消息已经接收到，但是没确认，是不会过期的。
public class TTLConsumer {private final static String QUEUE_NAME = "ttl_queue";public static void main(String[] argv) throws Exception {//创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");//创建频道，提供通信Connection connection = factory.newConnection();Channel channel = connection.createChannel();//指定消息队列的过期时间Map<String ,Object> args=new HashMap<>();args.put("x-message-ttl",5000);//args:指定参数channel.queueDeclare(QUEUE_NAME, false, false,false, args);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");//如何处理消息DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });}
}
2.1.2:给某条消息指定过期时间
//在发送者这边设置过期时间
public class TTLProducer {private final static String QUEUE_NAME = "ttl_queue";public static void main(String[] argv) throws Exception {//创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();//频道相当于客户端(jdbcClient,redisClient),提供了和消队列server建立通信，程序通过channel进行发送消息Channel channel = connection.createChannel()) {//创建消息队列，第二个参数(durable):是否开启持久化,第三个参数exclusiove:是否允许当前这个创建消息队列的//连接操作消息队列 第四个参数:没有人使用队列，是否需要删除String message = "Hello World!";//给消息指定过期时间AMQP.BasicProperties properties=new AMQP.BasicProperties.Builder().expiration("1000").build();channel.basicPublish("", QUEUE_NAME, properties, message.getBytes(StandardCharsets.UTF_8));System.out.println(" [x] Sent '" + message + "'");}}
2.2:死信队列
为了保证消息的可靠性，比如每条消息都成功消费，需要提供一个容错机制，即失败的消息怎么处理，相当于死信。
死信:过期的消息，拒收的消息，处理失败的消息，消息队列满了统称为死信。
死信队列:处理死信的队列。
死信交换机:给死信队列发送消息的交换机，也存在路由绑定。
a:创建死信交换机和死信队列
//声明死信交换机channel.exchangeDeclare(WORK_NAME,"direct");//声明死信队列String queueName="boss_queue";channel.queueDeclare(queueName,true,false,false,null);channel.queueBind(queueName,EXCHANGE_Name,"boss");String queueName2="waibao_queue";channel.queueDeclare(queueName2, false, false, false, null);channel.queueBind(queueName2,EXCHANGE_Name,"waibao");
b:给失败后的需要容错的队列绑定死信交换机
//声明交换机channel.exchangeDeclare(WORK_NAME, "direct");Map<String,Object> map=new HashMap<>();//声明要绑定的死信交换机map.put("x-dead-letter-exchange",DEAD_EXCHANGE_NAME);//声明要绑定的死信队列map.put("x-dead-letter-routing-key","waibao_queue");//创建消息队列String queueName="xiaodog_queue";channel.queueDeclare(queueName,true,false,false,map);channel.queueBind(queueName,WORK_NAME,"xiaodog");Map<String,Object> map2=new HashMap<>();//声明要绑定的死信交换机map2.put("x-dead-letter-exchange",DEAD_EXCHANGE_NAME);map2.put("x-dead-letter-routing-key","boss_queue");String queueName2="xiaocat_queue";channel.queueDeclare(queueName2,true,false,false,map2);channel.queueBind(queueName2,WORK_NAME,"xiaocat");
