Rabbitmq示例之点对点简单队列

上一篇 Rabbitmq基础知识
下一篇 >>>

---

1.特点

默认的传统队列是为均摊消费，存在不公平性；
如果每个消费者速度不一样的情况下，均摊消费是不公平的，应该是能者多劳。

2.程序代码示例

2.1依赖包

com.rabbitmqamqp-client3.6.5 

2.2 连接类

public class RabitMQConnection {

    public static Connection getConnection() throws IOException, TimeoutException {
        // 1.创建我们的连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 2.设置我们的连接地址
        connectionFactory.setHost("10.211.55.16");
        // 3.设置我们的端口号
        connectionFactory.setPort(5672);
        // 4.设置账号和密码
        connectionFactory.setUsername("jiang");
        connectionFactory.setPassword("123456");
        // 5.设置VirtualHost
        connectionFactory.setVirtualHost("/mytest1205");
        return connectionFactory.newConnection();
    }
}

2.3 生产者

public class Producer {
    private static final String QUEUE_NAME = "test1205";

    public static void main(String[] args) throws IOException, TimeoutException {
        System.out.println("生产者启动成功..");
        // 1.创建我们的连接
        Connection connection = RabitMQConnection.getConnection();
        // 2.创建我们通道
        Channel channel = connection.createChannel();
        for (int i = 0; i 

2.4 消费者

public class Consumer {
    private static final String QUEUE_NAME = "test1205";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.创建我们的连接
        Connection connection = RabitMQConnection.getConnection();
        // 2.创建我们通道
        Channel channel = connection.createChannel();
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "UTF-8");
                System.out.println("消费消息msg:" + msg);
            }
        };
        // 3.创建我们的监听的消息
        channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
   
    }
}

3.测试效果

消费者开多个时，默认是采用轮询(均摊)机制