Rabbitmq学习笔记（超详细）

#青青子衿#

377人浏览 · 2023-08-05 16:08:01

#青青子衿# · 2023-08-05 16:08:01 发布

RabbitMq

先了解一个概念

消息队列

存储消息的队列。

关键词：存储、消息、队列

存储：存数据

消息：某种数据结构，比如字符串、对象、二进制数据、json等等

队列：先进先出的数据结构

消息队列是特殊的数据库么？

可以这么理解。

应用场景（作用）：在多个不同的系统、应用之间实现消息的传输（也可以存储）。不需要考虑传输应用的编程语言、系统、框架等等。

可以让java开发的应用发消息，让php开发的应用收消息，这样就不用把所有代码写到同一个项日里（应用解耦)。

消息队列的模型

生产者：Producer,类比为快递员，发送消息的人（客户端）

消费者：Consumer,类比为取快递的人，接受读取消息的人（客户端）

消息：Message,类比为快递，就是生产者要传输给消费者的数据

消息队列：Queue

为什么不接传输，要用消息队列？生产者不用关心你的消费者要不要消费、什么时候消费，我只需要把东西给消息

队列，我的工作就算完成了。

生产者和消费者实现了解耦，互不影响。

三种应用场景

RabbitMq下载安装

先下载erlang

Otp 25.3.2 - Erlang/OTPThe official home of the Erlang Programming Languagehttps://www.erlang.org/patches/otp-25.3.2

再下载

RabbitMQ: easy to use, flexible messaging and streaming — RabbitMQhttps://www.rabbitmq.com/#getstarted

Installing on Windows — RabbitMQhttps://www.rabbitmq.com/install-windows.html

安装rabbitmg监控面板：

在rabbitmq安装目录的sbin中执行下述脚本：

rabbitmq-plugins.bat enable rabbitmq_management

访问：http:/localhost:15672,用户名密码都是guest

如果想要在远程服务器安装访问rabbitmq管理面板，你要自己创建一个管理员账号，不能用默认的guest,否侧

会被拦截（官方出于安全考虑）。

如果被栏截，可以自己创建管理员用户：

基本概念

4个角色

生产者：发消息的，通过交换机发送到各个队列

消费者：收消息的，从各个队列里取出消息

交换机：（exchange）将消息转发到各个队列里

队列：（queue）存放消息

基于AMPQ协议

AMQP，即Advanced Message Queuing Protocol（高级消息队列协议）

通俗来说，在异步通讯中，消息不会立刻到达接收方，而是被存放到一个容器中，当满足一定的条件之后，消息会被容器发送给接收方，这个容器即消息队列，而完成这个功能需要双方和容器以及其中的各个组件遵守统一的约定和规则，AMQP就是这样的一种协议，消息发送与接收的双方遵守这个协议可以实现异步通讯。这个协议约定了消息的格式和工作方式。

快速入门

RabbitMQ Tutorials — RabbitMQ

java文档

RabbitMQ tutorial - "Hello World!" — RabbitMQ

一对一发送消息

Helloword测试

https://github.com/rabbitmq/rabbitmq-tutorials/blob/main/java/Send.java

引入依赖，以后看官方文档都是这样

向这里就是让我们去Maven官方文档搜索com.rabbitmq

https://mvnrepository.com/

生产者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.nio.charset.StandardCharsets;

public class SingleProduce {

    private final static String QUEUE_NAME = "hello";

    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.queueDeclare(QUEUE_NAME, false, false, false, null);
            String message = "Hello World!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

执行完代码，面板有hello即为成功

消费者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;

public class SingleConsumer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
//        /创建连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //创建队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println("[*]Waiting for messages.To exit press CTRL+C");
        //定义了如何处理消息
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
            System.out.println("[x]Received '" + message + "'");
        };
        //消费消息，会持续阻塞
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {
        });
    }
}

生产一条消息立马就被消费了

源码解析

参数：

queueName:消息队列名称（注意，同名称的消息队列，只能用同样的参数创建一次）

durabale:消息队列重启后，消息是否丢失

exclusive:是否只允许当前这个创建消息队列的连接操作消息队列

autoDelete:没有人用队列后，是否要删除队列

多消费者

RabbitMQ tutorial - Work Queues — RabbitMQ

一个生产者给多台机器发消息

队列持久化

durable设置为Ture，服务器重启后队列不丢失

 channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

消息持久化， MessageProperties.PERSISTENT_TEXT_PLAIN,

channel.basicPublish("", TASK_QUEUE_NAME,
        MessageProperties.PERSISTENT_TEXT_PLAIN,
        message.getBytes("UTF-8"));

生产者代码

spackage com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;

import java.util.Scanner;

public class MultiProduce {

  private static final String TASK_QUEUE_NAME = "task_queue";

  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.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

        Scanner scanner=new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.nextLine();
            channel.basicPublish("", TASK_QUEUE_NAME,
                    MessageProperties.PERSISTENT_TEXT_PLAIN,
                    message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'");
        }

        
    }
  }

}

消费者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public class MultiCousumer {

    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        final Connection connection = factory.newConnection();
        for (int i = 0; i < 2; i++) {
            final Channel channel = connection.createChannel();

            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            channel.basicQos(1);
//定义了如何处理消息
            int finalI=i;
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");


                try {
                    System.out.println(" [x] Received '" +"编号"+finalI+":"+message + "'");
//                停20秒，模拟机器能力有限
                    Thread.sleep(20000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }finally {
                    System.out.println("[x] Done");
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
                }
            };

            channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {
            });
        }

        }

}

// 每一个消费者最多只能同时处理一个任务

channel.basicQos(1);

消息确认机制

为了保证消息被成功消费，rabbitmq提供了消息确认机制

ack：消息被消费

nack：消费失败

reject:拒绝

支持自动确认消息

 channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {
            });

指定确认某一条消息

channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);

默认写成false，怎么解释？（相当于你知道你快递站里面放了几个快递，一直没有去取，改成了Ture，它就自动帮你自动收货了，false，你需要你本人去确认收货）。

2个小技巧：

1.使用Scanner接受用户输入，便于快速发送多条消息

2.使用for循环创建多个消费者，便于快速验证队列模型工作机制

交换机

一个生产者给多个队列发消息

交换机：提供消息转发功能

fanout交换机

扇出、广播

程导航知

特点：消息会被转发到所有绑定到该交换机的队列

场景：很适用于发布订阅的场景。比如写日志，可以多个系统间共享交换机有多种类别：fanout、direct,topic,headers

生产者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


import java.util.Scanner;

public class FanoutProduce {

  private static final String EXCHANGE_NAME = "fanout-exchange";

  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, "fanout");
        Scanner scanner=new Scanner(System.in);
        while (scanner.hasNext()) {
            String message=scanner.nextLine();
            channel.basicPublish(EXCHANGE_NAME,"",null, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'");
        }

    }
  }
}

消费者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

public class FanoutConsumer {
  private static final String EXCHANGE_NAME = "fanout-exchange";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel1 = connection.createChannel();
    Channel channel2 = connection.createChannel();
//声明交换机
    channel1.exchangeDeclare(EXCHANGE_NAME, "fanout");
//      创建队列，随机分配一个队列名称
    String queueName1 ="小苗的工作队列";
    channel1.queueDeclare(queueName1, true, false, false, null);
//    绑定队列
    channel1.queueBind(queueName1, EXCHANGE_NAME, "");
//创建队列
    String queueName2 ="小刚的工作队列";
    channel1.queueDeclare(queueName2, true, false, false, null);
//绑定队列
    channel2.queueBind(queueName2, 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 '" + message + "'");
    };
    DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {
      String message = new String(delivery.getBody(), "UTF-8");
      System.out.println(" 小马 Received '" + message + "'");
    };
    channel1.basicConsume(queueName1, true, deliverCallback1, consumerTag -> { });
    channel2.basicConsume(queueName2, true, deliverCallback2, consumerTag -> { });
  }
}

Direct交换机

RabbitMQ tutorial - Routing — RabbitMQ

绑定：可以让交换机和队列进行关联，可以指定让交互机把什么样的消息发送给哪个队列（类似于计算机网络中，两个路由器，或者网络设备相互连接，也可以理解为网线)

routing Key:路由键，控制消息要转发给哪个队列的(IP地址)

特点：消息会根据路由键转发到指定的队列

场景：特定的消息只交给特定的系统（程序）来处理

绑定关系：完全匹配字符串

可以绑定同样的路由键。

比如发日志的场景，希望用独立的程序来处理不同级别的日志，比如C1系统处理error日志，C2系统处理其他级别的日志

生产者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.Scanner;

public class DirectProduce {

  private static final String EXCHANGE_NAME = "direct_exchange";

  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[] strings = userInput.split(" ");
            if (strings.length<1){
                continue;
        }
            String message=strings[0];
            String routingKey=strings[1];
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'with routingKey:'" + routingKey + "'");

        }


    }
  }
  //..
}

消费者代码

package com.zijin.mq;

import com.rabbitmq.client.*;

public class DirectConsumer {

  private static final String EXCHANGE_NAME = "direct_exchange";

  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, "direct");
      String queueName1 ="xiaoyu_queue";
      channel.queueDeclare(queueName1, true, false, false, null);
//    绑定队列
      channel.queueBind(queueName1, EXCHANGE_NAME, "xiaoyu");
      String queueName2 ="xiaochen_queue";
      channel.queueDeclare(queueName2, true, false, false, null);
//    绑定队列
      channel.queueBind(queueName2, EXCHANGE_NAME, "xiaochen");
    DeliverCallback xiaoyudeliverCallback = (consumerTag, delivery) -> {
        String message = new String(delivery.getBody(), "UTF-8");
        System.out.println(" [x] Received '" +
            delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
    };
      DeliverCallback xiaochendeliverCallback = (consumerTag, delivery) -> {
          String message = new String(delivery.getBody(), "UTF-8");
          System.out.println(" [x] Received '" +
                  delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
      };
    channel.basicConsume(queueName1, true, xiaoyudeliverCallback, consumerTag -> { });
      channel.basicConsume(queueName2, true, xiaochendeliverCallback, consumerTag -> { });
  }
}

topic交换机

RabbitMQ tutorial - Topics — RabbitMQ

特点：消息会根据一个模糊的路由键转发到指定的队列

场景：特定的一类消息可以交给特定的一类系统（程序）来处理

绑定关系：可以模糊匹配多个绑定

·*:匹配一个单词，比如*.orange,那么a.orange、b.orange都能匹配

·#:匹配0个或多个单词，比如a.#,那么a.a、a.b、a.a.a都能匹配

注意，这里的匹配和MySQL的k的%不一样，只能按照单词来匹配，每个'.分隔单词，如果是'#.'，其实可以忽略，匹配0个词也ok

老板要下发一个任务，让多个组来处理，这个下发消息的方式类比数据库中模糊查询的方式（%like%）

生产者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.Scanner;

public class TopicProduce {

    private static final String EXCHANGE_NAME = "topic_exchange";

    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[] strings = userInput.split(" ");
                if (strings.length < 1) {
                    continue;
                }
                String message = strings[0];
                String routingKey = strings[1];

                channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
                System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
            }
        }
        //..
    }
}

消费者代码

package com.zijin.mq;

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 = "topic_exchange";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();
//    创建队列
      String queueName1 ="frontend_queue";
      channel.queueDeclare(queueName1, true, false, false, null);
//    绑定队列
      channel.queueBind(queueName1, EXCHANGE_NAME, "#.前端.#");
      String queueName2 ="backend_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 xiaoadeliverCallback = (consumerTag, delivery) -> {
          String message = new String(delivery.getBody(), "UTF-8");
          System.out.println(" [xiaoa] Received '" +
                  delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
      };
      DeliverCallback xiaobdeliverCallback = (consumerTag, delivery) -> {
          String message = new String(delivery.getBody(), "UTF-8");
          System.out.println(" [xiaob] Received '" +
                  delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
      };
    DeliverCallback xiaocdeliverCallback = (consumerTag, delivery) -> {
        String message = new String(delivery.getBody(), "UTF-8");
        System.out.println(" [xiaoc] Received '" +
            delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
    };
    channel.basicConsume(queueName1, true, xiaoadeliverCallback, consumerTag -> { });
      channel.basicConsume(queueName2, true, xiaobdeliverCallback, consumerTag -> { });
      channel.basicConsume(queueName3, true, xiaocdeliverCallback, consumerTag -> { });
  }
}

结果

header交换机和RPC不推荐使用

核心特性

官方文档：Time-To-Live and Expiration — RabbitMQ

消息过期机制

可以给每条消息指定一个有效期，一段时间内未被消费者处理，就过期了。

示例场景：消费者（库存系统）挂了，一个订单15分钟还没被库存系统处理，这个订单其实已经失效了，哪怕库存系统再恢复，其实也不用扣减库存。

适用场景：清理过期数据、模拟延迟队列的实现（不开会员就慢速）、专门让某个程序处理过期请求

1)给队列中的所有消息指定过期时间

生产者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.nio.charset.StandardCharsets;

public class TtlProduce {


    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();
             Channel channel = connection.createChannel()) {
//            发送消息
            String message = "Hello World!";
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

消费者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;

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<String, Object>();
        args.put("x-message-ttl", 5000);
        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(), StandardCharsets.UTF_8);
            System.out.println("[x]Received '" + message + "'");
        };

        channel.basicConsume(QUEUE_NAME, flase, deliverCallback, consumerTag -> {
        });
    }
}

如果在过期时间内，还没有消费者取消息，消息才会过期。

注意，如果消息已经接收到，但是没确认，是不会过期的。（即使消息过期时间到了，后续依然会去处理，而不会被丢掉）

2）给某一条消息指定过期时间

AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
        .expiration("1000")
        .build();
channel.basicPublish("", "routing-key", properties, message.getBytes(StandardCharsets.UTF_8));

死信队列

官方文档：Dead Letter Exchanges — RabbitMQ

死信：过期的消息、拒收的消息、消息队列满了、处理失败的消息的统称

死信队列：专门处理死信的队列（注意，它就是一个普通队列，只不过是专门用来处理死信的，你甚至可以理解这个队列的名称叫“死信队列”)

死信交换机：专门给死信队列转发消息的交换机（注意，它就是一个普通交换机，只不过是专门给死信队列发消息而已，理解为这个交换机的名称就叫“死信交换机”)。也存在路由绑定

死信可以通过死信交换机绑定到死信队列。

生产者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.util.Scanner;

public class DlxDirectProduce {

  private static final String DEAD_EXCHANGE_NAME = "dlx_direct_exchange";
    private static final String EXCHANGE_NAME = "direct2_exchange";
  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(DEAD_EXCHANGE_NAME, "direct");
//        创建老板死信队列
        String queueName1 ="laoban_dlx_queue";
        channel.queueDeclare(queueName1, true, false, false, null);
//    绑定队列
        channel.queueBind(queueName1, DEAD_EXCHANGE_NAME, "laoban");
        //        创建外包死信队列
        String queueName2 ="waibao_dlx_queue";
        channel.queueDeclare(queueName2, true, false, false, null);
//    绑定队列
        channel.queueBind(queueName2, DEAD_EXCHANGE_NAME, "waibao");
        Scanner scanner=new Scanner(System.in);
        while (scanner.hasNext()) {
            String userInput = scanner.nextLine();
            String[] strings = userInput.split(" ");
            if (strings.length<1){
                continue;
        }
            String message=strings[0];
            String routingKey=strings[1];
            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'with routingKey:'" + routingKey + "'");

        }


    }
  }
  //..
}

消费者代码

package com.zijin.mq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.util.HashMap;
import java.util.Map;

public class DlxDirectConsumer {
  private static final String DEAD_EXCHANGE_NAME = "dlx_direct_exchange";
  private static final String EXCHANGE_NAME = "direct2_exchange";

  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, "direct");
//指定死信队列参数
    Map<String, Object> args1 = new HashMap<>();
//    指定要绑定到哪一个交换机
    args1.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
//    指定死信要转发到哪一个死信队列队列
    args1.put("x-dead-letter-routing-key", "waibao");
      String queueName1 ="xiaodog_queue";
      channel.queueDeclare(queueName1, true, false, false, args1);
//    绑定队列
      channel.queueBind(queueName1, EXCHANGE_NAME, "xiaodog");

    Map<String, Object> args2 = new HashMap<>();
    args2.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);
    args2.put("x-dead-letter-routing-key", "laoban");

      String queueName2 ="xiaocat_queue";
      channel.queueDeclare(queueName2, true, false, false, args2);
//    绑定队列
      channel.queueBind(queueName2, EXCHANGE_NAME, "xiaocat");
    DeliverCallback laobandeliverCallback = (consumerTag, delivery) -> {
        String message = new String(delivery.getBody(), "UTF-8");
//        拒绝消息
         channel.basicNack(delivery.getEnvelope().getDeliveryTag(),false,false);
        System.out.println(" [x] Received '" +
            delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
    };
      DeliverCallback waibaodeliverCallback = (consumerTag, delivery) -> {
          String message = new String(delivery.getBody(), "UTF-8");
//          拒绝消息
          channel.basicNack(delivery.getEnvelope().getDeliveryTag(),false,false);

          System.out.println(" [x] Received '" +
                  delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");
      };
      channel.basicConsume(queueName1, false, laobandeliverCallback, consumerTag -> { });
      channel.basicConsume(queueName2, false, waibaodeliverCallback, consumerTag -> { });
  }
}

RabbitMQ重点知识

也是面试考点

1.消息队列的概念、模型、应用场景

2.交换机的类别、路由绑定的关系

3.消息可靠性

a.消息确认机制(ack、nack、reject)

b.消息持久化(durable)

c.消息过期机制

d.死信队列

4.延迟队列（类似死信部队列）

5.顺序消费、消费幂等性（本次不讲）

6.可扩展性（仅作了解）

a.集群

b.故障的恢复机制

c.镜像

7.运维监控告警

RabbitMQ项目实战

看官方文档：Getting Started | Messaging with RabbitMQ（英文不好就不要了）

引入依赖

注意这里的版本一定要与自己的springboot版本一致

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
    <version>2.7.2</version>
</dependency>

在yml配置

rabbitmq:
  host: localhost
  port: 5672
  username: guest
  password: guest

创建交换机和队列

package com.zijin.bimq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class MqInitMain {
    public static void main(String[] args) {

//    创建测试所需要的交换机和队列，只需要执行一次
    try {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
//        创建交换机
        String EXCHANGE_NAME="code_exchange";

        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
//        创建队列
        String QUEUE_NAME="code_queue";
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);
        channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,"my_routingKey");
    } catch (Exception e) {


    }
    }
}

生产者代码

package com.zijin.bimq;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class MyMessageProducer {
    @Resource
  private RabbitTemplate rabbitTemplate;
  public void sendMessage(String exchange,String routingKey,String message){
      rabbitTemplate.convertAndSend(exchange,routingKey,message);
  }
}

消费者代码

package com.zijin.bimq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
@Slf4j
@Component

public class MyMessageConsumer {
    @RabbitListener(queues = {},ackMode = "MANUAL")
    @SneakyThrows
    public void receiveMessage(String message,Channel channel,@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag){
        log.info("receiveMessage message= {}",message);
        channel.basicAck(deliveryTag,false);
    }

 }

生产者测试类

package com.zijin.bimq;

import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;

import javax.annotation.Resource;

import static org.junit.jupiter.api.Assertions.*;
@SpringBootTest
class MyMessageProducerTest {
    @Resource
    private MyMessageProducer myMessageProducer;
    @Test
    void sendMessage(){
        myMessageProducer.sendMessage("code_exchange","my_routingKey","你好呀");
    }


}

天工造物局

北京人形旗下天工造物具身智能开源社区，聚焦具身天工与慧思开物两大平台

更多推荐

Handy 离线语音转写教程：免费开源，本地跑不连网

你肯定遇到过：语音输入 5 分钟，云转写服务要登录、要付费；或者出差没网，想口述一段会议纪要，发现全都没法用。Handy 离线语音转写工具就是为解决这个问题做的，声音全程不出你的电脑。## 🎯 这个工具是什么Handy 是一款免费、开源、跨平台（Windows / macOS / Linux）的离线语音转写应用。按住快捷键说话，松手后文字直接粘贴进任意输入框，全程不联网。核心能力就三条：

天工造物局

Supabase Edge Functions 编写与部署完整指南：Deno 运行时上的无服务器利器

Supabase Edge Functions 是基于 Deno 运行时编写、面向 Web 的无服务器函数服务，帮你把后端逻辑部署到全球边缘节点。作为 Postgres 开发平台的核心能力之一，它让你用 TypeScript 就能快速搭建低延迟的 API 端点、Webhook 接收器和 AI 集成服务，无需管理任何服务器。本文将从零带你完成一个 Edge Function 的创建、本地调试与全球部

天工造物局

celld从v0.1.0到v0.2.0为何不能滚动升级：两个破坏性变更详解

celld 是一个自托管、分布式的 Durable Objects 开源守护进程，它在你自己的机器上运行 Cloudflare Workers，每个对象对应一个独立的 SQLite 数据库，并持续复制到你自己拥有的 S3 兼容存储桶中。如果你正准备把 v0.1.0 的集群（fleet）升级到 v0.2.0，这篇文章必须先读：这次升级**不能滚动升级**，必须采用"先停光、再启动"的一次性发布方式。

天工造物局

所有评论(0)

查看更多评论

#青青子衿#

@m0_70638653

已为社区贡献1条内容

Rabbitmq学习笔记（超详细）

#青青子衿#

RabbitMq

RabbitMq下载安装

安装rabbitmg监控面板：

如果想要在远程服务器安装访问rabbitmq管理面板，你要自己创建一个管理员账号，不能用默认的guest,否侧

基本概念

快速入门

Helloword测试

引入依赖，以后看官方文档都是这样

生产者代码

消费者代码

源码解析

多消费者

生产者代码

消费者代码

消息确认机制

交换机

fanout交换机

生产者代码

消费者代码

Direct交换机

生产者代码

消费者代码

topic交换机

生产者代码

消费者代码

header交换机和RPC不推荐使用

核心特性

消息过期机制

1)给队列中的所有消息指定过期时间

2）给某一条消息指定过期时间

死信队列

生产者代码

消费者代码

RabbitMQ重点知识

RabbitMQ项目实战

引入依赖

在yml配置

创建交换机和队列

生产者代码

消费者代码

生产者测试类

所有评论(0)

温馨提示：您尚未绑定手机号

#青青子衿#