关于java:一个最简单的消息队列带你理解-RabbitMQ

30次阅读

共计 7459 个字符，预计需要花费 19 分钟才能阅读完成。

作者：海向 \
起源：https://www.cnblogs.com/haixi…

RabbitMQ 简述

RabbitMQ 是一个音讯代理：它承受并转发音讯。您能够将其视为邮局：当您将要把寄发的邮件投递到邮箱中时，您能够确信 Postman 学生最终会将邮件发送给收件人。在这个比喻中，RabbitMQ 是一个邮箱，邮局和邮递员，用来承受，存储和转发二进制数据块的音讯。

队列就像是在 RabbitMQ 中表演邮箱的角色。尽管音讯通过 RabbitMQ 和应用程序，但它们只能存储在队列中。队列只受主机的内存和磁盘限度的限度，它实质上是一个大的音讯缓冲区。许多生产者能够发送到一个队列的音讯，许多消费者能够尝试从一个队列接收数据。

producer 即为生产者，用来产生音讯发送给队列。consumer 是消费者，须要去读队列内的音讯。producer，consumer 和 broker(rabbitMQ server)不用驻留在同一个主机上; 的确在大多数应用程序中它们是这样散布的。

简略队列

简略队列是最简略的一种模式，由生产者、队列、消费者组成。生产者将音讯发送给队列，消费者从队列中读取音讯实现生产。

在下图中，“P”是咱们的生产者，“C”是咱们的消费者。两头的框是队列 – RabbitMQ 代表消费者的音讯缓冲区。

java 形式

生产者

package com.anqi.mq.nat;

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

public class MyProducer {
    private static final String QUEUE_NAME = "ITEM_QUEUE";

    public static void main(String[] args) throws Exception {
        //1. 创立一个 ConnectionFactory 并进行设置
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");

        //2. 通过连贯工厂来创立连贯
        Connection connection = factory.newConnection();

        //3. 通过 Connection 来创立 Channel
        Channel channel = connection.createChannel();

        // 理论场景中，音讯多为 json 格局的对象
        String msg = "hello";
        //4. 发送三条数据
        for (int i = 1; i <= 3 ; i++) {channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            System.out.println("Send message" + i +":" + msg);
        }

        //5. 敞开连贯
        channel.close();
        connection.close();}
}
/**
 * Declare a queue
 * @param queue the name of the queue
 * @param durable true if we are declaring a durable queue (the queue will survive a server restart)
 * @param exclusive true if we are declaring an exclusive queue (restricted to this connection)
 * @param autoDelete true if we are declaring an autodelete queue (server will delete it when no longer in use)
 * @param arguments other properties (construction arguments) for the queue
 * @return a declaration-confirm method to indicate the queue was successfully declared
 * @throws java.io.IOException if an error is encountered
 */
Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments) throws IOException;

/**
 * Publish a message
 * @see com.rabbitmq.client.AMQP.Basic.Publish
 * @param exchange the exchange to publish the message to
 * @param routingKey the routing key
 * @param props other properties for the message - routing headers etc
 * @param body the message body
 * @throws java.io.IOException if an error is encountered
 */
void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;


/**
 * Start a non-nolocal, non-exclusive consumer, with
 * a server-generated consumerTag.
 * @param queue the name of the queue
 * @param autoAck true if the server should consider messages
 * acknowledged once delivered; false if the server should expect
 * explicit acknowledgements
 * @param callback an interface to the consumer object
 * @return the consumerTag generated by the server
 * @throws java.io.IOException if an error is encountered
 * @see com.rabbitmq.client.AMQP.Basic.Consume
 * @see com.rabbitmq.client.AMQP.Basic.ConsumeOk
 * @see #basicConsume(String, boolean, String, boolean, boolean, Map, Consumer)
 */
String basicConsume(String queue, boolean autoAck, Consumer callback) throws IOException;

消费者

package com.anqi.mq.nat;

import com.rabbitmq.client.*;
import java.io.IOException;

public class MyConsumer {

    private static final String QUEUE_NAME = "ITEM_QUEUE";

    public static void main(String[] args) throws Exception {
        //1. 创立一个 ConnectionFactory 并进行设置
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");

        //2. 通过连贯工厂来创立连贯
        Connection connection = factory.newConnection();

        //3. 通过 Connection 来创立 Channel
        Channel channel = connection.createChannel();

        //4. 申明一个队列
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        System.out.println("[*] Waiting for messages. To exit press CTRL+C");

        /*
           true: 示意主动确认，只有音讯从队列中获取，无论消费者获取到音讯后是否胜利生产，都会认为音讯曾经胜利生产
           false: 示意手动确认，消费者获取音讯后，服务器会将该音讯标记为不可用状态，期待消费者的反馈，如果消费者一
           直没有反馈，那么该音讯将始终处于不可用状态，并且服务器会认为该消费者曾经挂掉，不会再给其发送音讯，直到该消费者反馈。*/

        //5. 创立消费者并接管音讯
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {String message = new String(body, "UTF-8");
                System.out.println("[x] Received'" + message + "'");
            }
        };

        //6. 设置 Channel 消费者绑定队列
        channel.basicConsume(QUEUE_NAME, true, consumer);

    }
}

Send message1 : hello
Send message2 : hello
Send message3 : hello

 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'hello'
 [x] Received 'hello'
 [x] Received 'hello'

当咱们启动生产者之后查看 RabbitMQ 治理后盾能够看到有一条音讯正在期待被生产。

当咱们启动消费者之后再次查看，能够看到积压的一条音讯曾经被生产。

总结

队列申明 queueDeclare 的参数：第一个参数示意队列名称、第二个参数为是否长久化（true 示意是，队列将在服务器重启时生存）、第三个参数为是否是独占队列（创建者能够应用的公有队列，断开后主动删除）、第四个参数为当所有消费者客户端连贯断开时是否主动删除队列、第五个参数为队列的其余参数。

basicConsume 的第二个参数 autoAck: 应答模式，true：自动应答，即消费者获取到音讯，该音讯就会从队列中删除掉，false：手动应答，当从队列中取出音讯后，须要程序员手动调用办法应答，如果没有应答，该音讯还会再放进队列中，就会呈现该音讯始终没有被生产掉的景象。

这种简略队列的模式，零碎会为每个队列隐式地绑定一个默认交换机，交换机名称为 ” (AMQP default)”，类型为直连 direct，当你手动创立一个队列时，零碎会主动将这个队列绑定到一个名称为空的 Direct 类型的交换机上，绑定的路由键 routing key 与队列名称雷同，相当于 channel.queueBind(queue:"QUEUE_NAME", exchange:"(AMQP default)“, routingKey:"QUEUE_NAME"); 尽管实例没有显式申明交换机，然而当路由键和队列名称一样时，就会将音讯发送到这个默认的交换机中。这种形式比较简单，然而无奈满足简单的业务需要，所以通常在生产环境中很少应用这种形式。

The default exchange is implicitly bound to every queue, with a routing key equal to the queue name. It is not possible to explicitly bind to, or unbind from the default exchange. It also cannot be deleted. 默认交换机隐式绑定到每个队列，其中路由键等于队列名称。不可能显式绑定到，或从缺省替换中解除绑定。它也不能被删除。

——引自 RabbitMQ 官网文档

spring-amqp 形式

引入 Maven 依赖

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.6.0</version>
</dependency>        
<dependency>
    <groupId>org.springframework.amqp</groupId>
    <artifactId>spring-rabbit</artifactId>
    <version>2.1.5.RELEASE</version>
</dependency>

spring 配置文件

<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/rabbit
           https://www.springframework.org/schema/rabbit/spring-rabbit.xsd
           http://www.springframework.org/schema/beans
           https://www.springframework.org/schema/beans/spring-beans.xsd">

    <rabbit:connection-factory id="connectionFactory" host="localhost" virtual-host="/"
    username="guest" password="guest"/>
    <rabbit:template id="amqpTemplate" connection-factory="connectionFactory"/>
    <rabbit:admin connection-factory="connectionFactory"/>
    <rabbit:queue name="MY-QUEUE"/>
</beans>

应用测试

import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

public class Main {public static void main(String[] args) {ApplicationContext app = new ClassPathXmlApplicationContext("spring/rabbit-context.xml");
        AmqpTemplate amqpTemplate = app.getBean(AmqpTemplate.class);
        amqpTemplate.convertAndSend("MY-QUEUE", "Item");
        String msg = (String) amqpTemplate.receiveAndConvert("MY-QUEUE");
        System.out.println(msg);
    }
}

参考办法

/**
 * Convert a Java object to an Amqp {@link Message} and send it to a specific exchange
 * with a specific routing key.
 *
 * @param exchange the name of the exchange
 * @param routingKey the routing key
 * @param message a message to send
 * @throws AmqpException if there is a problem
 */
void convertAndSend(String exchange, String routingKey, Object message) throws AmqpException;
/**
     * Receive a message if there is one from a specific queue and convert it to a Java
     * object. Returns immediately, possibly with a null value.
     *
     * @param queueName the name of the queue to poll
     * @return a message or null if there is none waiting
     * @throws AmqpException if there is a problem
     */
@Nullable
Object receiveAndConvert(String queueName) throws AmqpException;

近期热文举荐：

1.1,000+ 道 Java 面试题及答案整顿(2021 最新版)

2. 终于靠开源我的项目弄到 IntelliJ IDEA 激活码了，真香！

3. 阿里 Mock 工具正式开源，干掉市面上所有 Mock 工具！

4.Spring Cloud 2020.0.0 正式公布，全新颠覆性版本！

5.《Java 开发手册（嵩山版）》最新公布，速速下载！

感觉不错，别忘了顺手点赞 + 转发哦！

正文完
java
发表至： java
2021-07-06
 0
关于java:CrashSight-常规功能特色功能介绍
关于java:kafka-原理
关于java:Java-18-新特性使用Java代码启动jwebserver
存疑-JVMCFRE003-bad-major-version-offset6