代码探险家

这个屌丝很懒，什么也没留下！

热门标签

RabbitMQ详解（二）五种队列及其实现_rabbitmq发送消息给queue

作者：代码探险家 | 2024-07-11 04:33:52

踩

rabbitmq发送消息给queue

1. 简单模式

一个生产者P发送消息到队列Q，一个消费者C接收。

1.1 代码示例

pom文件


<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.6.0</version>
</dependency>

RabbitMQUtil工具类


package util;
 
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * RabbitMQ Util
 *
 * @author HarryZhang
 */
public class RabbitMQUtil {
    private static Connection connection;
 
    /**
     * getConnection
     *
     * @return Connection
     */
    public static Connection getConnection() {
        if (connection != null) {
            return connection;
        }
        ConnectionFactory factory = new ConnectionFactory();
 
        factory.setHost("127.0.0.1");//MQ ip address
 
        factory.setPort(5672);//port
 
        factory.setUsername("HarryZhang");//username
 
        factory.setPassword("123456");//password
 
        try {
            connection = factory.newConnection();
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
        return connection;
    }
}

消息生产者


import com.rabbitmq.client.Channel;
import util.RabbitMQUtil;
 
import java.io.IOException;
 
public class Producer {
    //队列名
    private static final String QUEUE_NAME="debugers_test";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtil.getConnection().createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        String message="Hello HarryZhang";
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
        System.out.println("推送成功");
    }
}

消息消费者


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import util.RabbitMQUtil;
 
import java.io.IOException;
 
public class Consumer {
    private static final String QUEUE_NAME="debugers_test";
    public static void main(String[] args) throws IOException {
        Channel channel = RabbitMQUtil.getConnection().createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        DeliverCallback deliverCallback=(consumerTag, delivery) -> {
            String message=new String(delivery.getBody(),"utf-8");
            System.out.println("接收消息:"+message);
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
    }
}

2. 工作队列（Work Queues）

一个生产者，多个消费者，一个消息只能被一个消费者获取。多个消费者只有一个队列。

python-two

2.1 轮询分发（Round Robin）

使用工作队列的优点之一就是可以轻易的并行工作。如果我们积压了好多工作，我们可以通过增加工作者（消费者）来解决这一问题，使得系统的伸缩性更加容易。在默认情况下，RabbitMQ采用轮询分发策略将逐个发送消息到在序列中的下一个消费者(而不考虑每个任务的时长等等，且是提前一次性分配，并非一个一个分配)。平均每个消费者获得相同数量的消息。

消费者


package worker;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import util.RabbitMQUtil;
 
public class Recv1 {
    private static final String QUEUE_NAME="debugers_test";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        DeliverCallback deliverCallback=(consumerTag, delivery) -> {
            String message=new String(delivery.getBody(),"utf-8");
            System.out.println(" [x] Received '" + message + "'");
            // 休眠1秒
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
    }
}

生产者


package workerfair;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import util.RabbitMQUtil;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
 
public class Producer {
    private static final String QUEUE_NAME="debugers_test";
    public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        for (int i = 0; i < 100; i++) {
            // 消息内容
            String message = "" + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
 
            Thread.sleep(i * 10);
        }
        channel.close();
        connection.close();
    }
}

2.2 公平分发（Fair Dispatch）

有可能消费者处理消息的能力有差异（硬件设备，网络原因），我们期望处理能力强的消费者多处理消息，处理能力弱的消费者少处理消息。通过basicQos(perfetch)和autoAck配合也可以实现。

basicQos：设置同一时刻服务器只会发perfetch**(此处为1)**条消息给消费者
autoAck：将自动应答改为手动。就处理完一条消息后手动提交。

消费者


package workerfair;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import util.RabbitMQUtil;
 
public class Recv1 {
    private static final String QUEUE_NAME="debugers_test";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        DeliverCallback deliverCallback=(consumerTag, delivery) -> {
            String message=new String(delivery.getBody(),"utf-8");
            System.out.println(" [x] Received '" + message + "'");
            // 休眠1秒
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }finally {
                System.out.println("[x] Done");
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
            }
        };
        //修改为手动应答，true为自动应答,false相反
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });
    }
}

注意：使用公平分发，必须关闭自动应答ack，然后改成手动应答方式。

3. 发布订阅（Publish/Subscribe）

一个生产者发送的消息可能会被多个消费者获取。一个生产者、一个交换机、多个队列、多个消费者。

注：X表示交换器，在RabbitMQ中，交换器主要有四种类型:direct、fanout、topic、headers。这里采用的是fanout类型。后面会详细介绍这几种交换器。

3.1 模式特点

一个生产者，多个消费者
每个消费者都有自己的队列
生产者没有直接将消息发送到队列，而是发送到交换机（Exchange）
每个队列都要绑定交换机
生产者发送到消息经过交换机 --> 到达队列–> 可以实现一个消息被多个消费者消

3.2 代码示例

生产者


package ps;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import util.RabbitMQUtil;
 
public class Send {
    private final static String EXCHANGE_NAME = "test_exchange_fanout";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
        // 消息内容
        String message = "Hello World!";
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
        //关闭通道连接
        channel.close();
        connection.close();
    }
}

注意： 此时的交换机是没有绑定队列的，发送消息肯定会丢失。交换机本身是不能存储数据的，队列才是存储数据的

消费者


package ps;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import util.RabbitMQUtil;
 
public class Recv1 {
    private final static String EXCHANGE_NAME = "test_exchange_fanout";
    private final static String QUEUE_NAME="debugers_test_sms";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "utf-8");
            System.out.println(" [x] Received '" + message + "'");
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        //修改为手动应答，true为自动应答,false相反
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
        });
    }
}


package ps;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import util.RabbitMQUtil;
 
public class Recv2 {
    private final static String EXCHANGE_NAME = "test_exchange_fanout";
    private final static String QUEUE_NAME="debugers_test_email";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "utf-8");
            System.out.println(" [x] Received '" + message + "'");
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        //修改为手动应答，true为自动应答,false相反
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
        });
    }
}

队列和交换机绑定关系

1563099102615

4. 路由模式（Routing）

生产者将消息发送到direct交换器，在绑定队列和交换器的时候有一个路由key，生产者发送的消息会指定一个路由key，那么消息只会发送到相应key相同的队列，接着监听该队列的消费者消费消息。也就是让消费者有选择性的接收消息。

4.1 代码示例

绑定

绑定(binding)是队列和交换机之间的关系。你可以简单的理解为：队列对来自此交换机的消息感兴趣。

绑定可以采用额外的routingKey参数。千万别和basicPublish混淆了。


//绑定队列到交换机
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");

生产者


package routing;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import util.RabbitMQUtil;
 
public class Send {
    private final static String EXCHANGE_NAME = "test_exchange_direct";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME,"direct");
        // 消息内容
        String message = "Hello World!";
        //参数2为routingKey
        channel.basicPublish(EXCHANGE_NAME, "delete", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
        //关闭通道连接
        channel.close();
        connection.close();
    }
}

消费者1


package routing;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import util.RabbitMQUtil;
 
public class Recv1 {
    private final static String EXCHANGE_NAME = "test_exchange_direct";
    private static final String QUEUE_NAME="debugers_test_sms";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "utf-8");
            System.out.println(" [x] Received '" + message + "'");
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        //修改为手动应答，true为自动应答,false相反
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
        });
    }
}

消费者2


package routing;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import util.RabbitMQUtil;
 
public class Recv2 {
    private final static String EXCHANGE_NAME = "test_exchange_direct";
    private static final String QUEUE_NAME="debugers_test_email";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "utf-8");
            System.out.println(" [x] Received '" + message + "'");
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        //修改为手动应答，true为自动应答,false相反
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
        });
    }
}

5. 主题模式（topic）

上面的路由模式是根据路由key进行完整的匹配（完全相等才发送消息），这里的通配符模式通俗的来讲就是模糊匹配。

发送到topic交换的消息不能具有任意的 routing_key- 它必须是由点(.)分隔的单词列表。单词可以是任何内容，但通常它们指定与消息相关的一些功能。

一些有效的路由键示例：stock.usd.nyse，nyse.vmw，quick.orange.rabbit。路由密钥中可以包含任意数量的字符，最多可达255个字节。

绑定键有两个重要的特殊特性：（是用.分割的单词，而不是字符）

*：可以替代一个单词。
#：可以替换零个或多个单词。

5.1 代码示例

生产者


package topic;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import util.RabbitMQUtil;
 
public class Send {
    private final static String EXCHANGE_NAME = "test_exchange_topic";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME,"topic");
        // 消息内容
        String message = "Hello World!";
        //参数2为routingKey
        channel.basicPublish(EXCHANGE_NAME, "routeKey.a", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
        //关闭通道连接
        channel.close();
        connection.close();
    }
}

消费者1


package topic;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import util.RabbitMQUtil;
 
public class Recv1 {
    private final static String EXCHANGE_NAME = "test_exchange_topic";
    private static final String QUEUE_NAME="debugers_test_sms";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "utf-8");
            System.out.println(" [x] Received '" + message + "'");
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        //修改为手动应答，true为自动应答,false相反
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
        });
    }
}

消费者2


package topic;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DeliverCallback;
import util.RabbitMQUtil;
 
public class Recv2 {
    private final static String EXCHANGE_NAME = "test_exchange_topic";
    private static final String QUEUE_NAME="debugers_test_email";
    public static void main(String[] args) throws Exception {
        Connection connection = RabbitMQUtil.getConnection();
        Channel channel = connection.createChannel();
        //声明队列
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "routeKey.*");
        // 同一时刻服务器只会发一条消息给消费者
        channel.basicQos(1);
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "utf-8");
            System.out.println(" [x] Received '" + message + "'");
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        //修改为手动应答，true为自动应答,false相反
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {
        });
    }
}

6. 四种交换器（Exchange）

前面五种队列模式介绍完了，但是实际上只有三种，第一种简单队列，第二种工作模式，剩下的三种都是和交换器绑定的合起来称为一种，这小节我们就来详细介绍交换器。

交换器分为四种，分别是：direct、fanout、topic和 headers。

前面三种分别对应路由模式、发布订阅模式和通配符模式，headers 交换器允许匹配 AMQP 消息的 header 而非路由键，除此之外，header 交换器和 direct 交换器完全一致，但是性能却差很多，因此基本上不会用到该交换器，这里也不详细介绍。

6.1 direct

如果路由键完全匹配的话，消息才会被投放到相应的队列。

6.2 fanout

当发送一条消息到fanout交换器上时，它会把消息投放到所有附加在此交换器上的队列。

6.3 topic

设置模糊的绑定方式，“*”操作符将“.”视为分隔符，匹配单个字符；“#”操作符没有分块的概念，它将任意“.”均视为关键字的匹配部分，能够匹配多个字符。

7. 总结

关于 RabbitMQ 的五种队列，其实实际使用最多的是最后一种主题模式，通过模糊匹配，使得操作更加自如。那么我们总结一下有交换器参与的队列（最后三种队列）工作方式如下：

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/代码探险家/article/detail/808765