分布式消息队列:Rabbitmq(2)-CSDN博客

阿里云国内75折 回扣 微信号：monov8
阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6

目录

一:交换机

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;
                    }
                    //指定路由key
                    String 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;
                    }
                    //指定路由key
                    String 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;
                }
                //指定路由key
                String 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");

阿里云国内75折 回扣 微信号：monov8
阿里云国际，腾讯云国际，低至75折。AWS 93折 免费开户实名账号 代冲值 优惠多多 微信号：monov8 飞机：@monov6
标签: RabbitMQ
返回列表

上一篇：rust 创建多线程web server-CSDN博客

下一篇：【计算机毕设小程序案例】基于微信小程序的图书馆座位预定系统-CSDN博客

“分布式消息队列:Rabbitmq(2)-CSDN博客” 的相关文章

RabbitMQ之消息转换器1年前 (2023-02-02)
RabbitMQ-客户端源码之AMQChannel1年前 (2023-02-04)
【RabbitMQ】Fanout、Direct、Topic、消息转换器1年前 (2023-02-07)
linux安装rabbitmq1年前 (2023-02-08)
RabbitMQ-客户端源码之AMQChannel1年前 (2023-02-08)
RabbitMQ的安装与基本使用（windows+php版本）1年前 (2023-02-09)
linux CentOS 7 安装rabbitMq1年前 (2023-02-09)
服务异步通信 RabbitMQ-高级篇1年前 (2023-02-09)
RabbitMQ1年前 (2023-02-09)
RabbitMq消息转换器1年前 (2023-02-09)

阿里云国际版