RabbitMQ-精简版-Toy模板网

这篇具有很好参考价值的文章主要介绍了RabbitMQ-精简版。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

声明：本博客已标明出处，如有侵权请告知，马上删除。

RabbitMQ 相关概述：

RabbitMQ是微服务间实现异步通讯的一种技术.

RabbitMQ中的一些角色：

publisher：生产者

consumer：消费者

exchange个：交换机，负责消息路由

queue：队列，存储消息

virtualHost：虚拟主机，隔离不同租户的exchange、queue、消息的隔离

入门案例

生产者->发消息

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.136.134");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("jd");
        factory.setPassword("123321");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功：【" + message + "】");

        // 5.关闭通道和连接
        channel.close();
        connection.close();

    }
}

消费者->收消息

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.136.134");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("jd");
        factory.setPassword("123321");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

        // 2.创建通道Channel
        Channel channel = connection.createChannel();

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息：【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}

SpringAMQP-简单消息

SpringAMQP是基于RabbitMQ封装的一套模板，并且还利用SpringBoot对其实现了自动装配，使用起来非常方便。

SpringAmqp的官方地址：https://spring.io/projects/spring-amqp

SpringAMQP提供了三个功能：

自动声明队列、交换机及其绑定关系

基于注解的监听器模式，异步接收消息

封装了RabbitTemplate工具，用于发送消息

SpringAMQP的简单消息没有交换机角色.

准备工作

1.导入坐标

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

2.配置地址

spring:
  rabbitmq:
    host: 192.168.136.134 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: jd # 用户名
    password: 123321 # 密码

SimpleQueue

简单消息: 一个队列对应一个消费者

生产者-发消息

@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSimpleQueue() {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, spring amqp!";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

消费者-收消息

@Component
public class SpringRabbitListener {
	
    //queuesToDeclare: 队列名称,如果队列不存在,则会自动创建
    @RabbitListener(queuesToDeclare = @Queue("simple.queue"))
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("spring 消费者接收到消息：【" + msg + "】");
    }
}

WorkQueue

Work queues，也被称为（Task queues），任务模型。简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。

生产者-发消息

@Test
public void testWorkQueue() throws InterruptedException {
    // 队列名称
    String queueName = "simple.queue";
    // 消息
    String message = "hello, message_";
    for (int i = 0; i < 50; i++) {
        // 发送消息---> 唯一ID
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}

消费者-收消息

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(20);
}

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);
}

SpringAMQP-发布/订阅

在订阅模型中，多了一个exchange角色：

Publisher：生产者，也就是要发送消息的程序，但是不再发送到队列中，而是发给X（交换机）

Exchange：交换机，一方面，接收生产者发送的消息。另一方面，决定如何处理消息。Exchange有以下3种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列

Direct：定向，把消息交给符合指定routing key 的队列

Topic：通配符，把消息交给符合routing pattern（路由模式）的队列

Consumer：消费者，与以前一样，订阅队列，没有变化

Queue：消息队列也与以前一样，接收消息、缓存消息。

注意:

Exchange只负责转发消息，不具备存储消息的能力

如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

声明队列和交换机

基于@Bean配置

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
 * 声明"队列"和"交换机"
 * 交换机的类型可以切换: FanoutExchange,DirectExchange,TopicExchange.
 */
@Configuration
public class FanoutConfig {
    
    /**
     * 声明交换机-jd.fanout
     * 交换机的类型: FanoutExchange
     * @return
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("jd.fanout");
    }

    /**
     * 声明队列-fanout.queue1
     * @return
     */
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout.queue1");
    }

    /**
     * 绑定队列1到交换机
     * @param fanoutQueue1 队列1,参数名字必须和fanoutQueue1的方法名保持一致(原因在于bean的名字)
     * @param fanoutExchange 交换机,参数名必须和fanoutExchange方法名保持一致
     * @return
     */
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }

    /**
     * 声明队列-fanout.queue2
     * @return
     */
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout.queue2");
    }

    /**
     * 绑定队列2到交换机
     * @param fanoutQueue2 队列2,参数名字必须和fanoutQueue2的方法名保持一致(原因在于bean的名字)
     * @param fanoutExchange 交换机,参数名必须和fanoutExchange方法名保持一致
     * @return
     */
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange) {
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }
}

基于注解配置

// 在消息的消费方上,直接绑定
// 通过@RabbitListener注解, 直接定义"交换机"和"队列", 并绑定他们的关系
@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "队列名"),
    exchange = @Exchange(name = "交换机名", type = ExchangeTypes.交换机类型)
))

FanoutExchangeQueue

概述:

Fanout：广播，将消息交给所有绑定到交换机的队列.

假设:

已通过"@Bean配置"完成了"队列"和"交换机"的配置.

交换机名:  jd.fanout

队列名: fanout.queue1,  fanout.queue2

生产者-发消息

@Test
public void testSendFanoutExchange() {
    // 交换机名称
    String exchangeName = "jd.fanout";
    // 消息
    String message = "hello, every one!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

消费者-收消息

/**
 * FanoutQueue: 可以理解为是广播模式
 * 所有和Fanout交换机绑定的消息队列,都会接受到交换机中的所有消息
 */
@Component
public class FanoutQueueListener {

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) {
        System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】");
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg) {
        System.out.println("消费者接收到fanout.queue2的消息：【" + msg + "】");
    }

}

DirectExchangeQueue

概述:

要根据Routing Key进行判断，队列的Routingkey与消息的 Routing key完全一致时，才会接收到消息

假设:

没有配置"交换机"和"队列"信息

生产者-发消息

@Test
public void testSendDirectExchange() {
    // 交换机名称
    String exchangeName = "jd.direct";
    // 消息
    String message = "hello, red!";
    // 发送消息(第二个参数就是RoutingKey)
    rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

消费者-收消息

/**
 * 根据key获取要消费的消息
 */
@Component
public class DirectQueueListener {

    //定义"交换机-jd.direct"
    //定义"队列-direct.queue1"
    //绑定"交换机"和"队列"
    //绑定key. 
    //要求: 消息的RoutingKey是red或blue,才执行
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1"),
            exchange = @Exchange(name = "jd.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "blue"}
    ))
    public void listenDirectQueue1(String msg){
        System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】");
    }

    //要求: 消息的RoutingKey是red或yellow,才执行
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2"),
            exchange = @Exchange(name = "jd.direct", type = ExchangeTypes.DIRECT),
            key = {"red", "yellow"}
    ))
    public void listenDirectQueue2(String msg){
        System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】");
    }
}

TopicExchangeQueue

概述:

Routing key 可以使用通配符的DirectExchangeQuery.

通配符:

#：匹配一个或多个词

*：匹配不多不少恰好1个词

生产者-发消息

@Test
public void testSendTopicExchange() {
    // 交换机名称
    String exchangeName = "jd.topic";
    // 消息
    String message = "今天天气不错，我的心情好极了!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.weather", message);
}

消费者-收消息

/**
 * 支持通配符的FanoutQueue
 */
@Component
public class TopicQueueListener {

    //定义"交换机-jd.topic"
    //定义"队列-topic.queue1"
    //绑定"交换机"和"队列"
    //绑定key. 
    //要求: 消息的RoutingKey是以china.开头的,才执行
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue1"),
            exchange = @Exchange(name = "jd.topic", type = ExchangeTypes.TOPIC),
            key = "china.#"
    ))
    public void listenTopicQueue1(String msg){
        System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】");
    }
    
    

    //要求: 消息的RoutingKey是以.news开头的,才执行
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "topic.queue2"),
            exchange = @Exchange(name = "jd.topic", type = ExchangeTypes.TOPIC),
            key = "#.news"
    ))
    public void listenTopicQueue2(String msg){
        System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】");
    }
}

消息转换器

Spring会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。

默认情况下Spring采用的序列化方式是JDK序列化. 我们可以修改其默认转换方式文章来源地址https://www.toymoban.com/news/detail-414158.html

添加依赖

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>

配置转换器

@Bean
public MessageConverter jsonMessageConverter(){
    return new Jackson2JsonMessageConverter();
}

到了这里，关于RabbitMQ-精简版的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

RabbitMQ-精简版

RabbitMQ 相关概述：

入门案例

生产者->发消息

消费者->收消息

SpringAMQP-简单消息

准备工作

SimpleQueue

生产者-发消息

消费者-收消息

WorkQueue

生产者-发消息

消费者-收消息

相关补充

SpringAMQP-发布/订阅

声明队列和交换机

基于@Bean配置

基于注解配置

FanoutExchangeQueue

生产者-发消息

消费者-收消息

DirectExchangeQueue

生产者-发消息

消费者-收消息

TopicExchangeQueue

生产者-发消息

消费者-收消息

消息转换器

添加依赖

配置转换器

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

支付宝扫一扫领取红包，优惠每天领

二维码1

二维码2