RabbitMQ使用详解_大数据系统

一、消息中间件介绍

自行百度

作用：

异步处理
应用解耦
流量削峰

二、安装

Windows和Linux版需要安装Erlang和RabbitMQ

启动：rabbitmq-server.bat

自行百度

三、控制台使用

端口：15672和5672，超级管理员账号/密码：guest/guest

3.1 添加用户 3.2 Virtual Hosts

RabbitMq的VirtualHost（虚拟消息服务器），每个VirtualHost相当于一个相对独立的RabbitMQ服务器；每个VirtualHost之间是相互隔离的，exchange、queue、message不能互通。 Virtual Hosts相当于MySQL中的一个库

3.3 添加Virtual Hosts

VirtualHosts中Name一般以/开头

添加完Virtualhosts之后，如上图，红框可以点击，点击后进入下图所示，可以进行virtualhosts授权

四、操作队列 4.1 simple简单队列 4.1.1 模型

生产者 --> 队列 --> 消费者

4.1.2 获取MQ连接

引入依赖


    
        com.rabbitmq
        amqp-client
        5.7.0
    
    
        org.slf4j
        slf4j-api
        1.7.30
    
    
        org.slf4j
        slf4j-log4j12
        1.7.30
        test
    
    
        log4j
        log4j
        1.2.17
    
    
        junit
        junit
        4.12

定义连接MQ的工具类

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConnectionUtils {
    // 获取MQ的连接
    public static Connection getConn() throws IOException, TimeoutException {
        // 定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置服务地址
        factory.setHost("127.0.0.1");
        // 设置端口
        factory.setPort(5672);
        // 设置Virtual Hosts
        factory.setVirtualHost("/first");
        // 设置用户名
        factory.setUsername("admin");
        // 设置密码
        factory.setPassword("admin");
        return factory.newConnection();
    }
}

发送者（生产者）

public class Send {
    private static final String QUEUE_NAME = "simple_queue_test";
    
    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection conn = ConnectionUtils.getConn();
        // 创建通道
        Channel channel = conn.createChannel();
        // 创建队列声明
        
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String msg = "hello world!!!";
        
        channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
        channel.close();
        conn.close();
    }
}

接受者（消费者）

public class Receive {
    private static final String QUEUE_NAME = "simple_queue_test";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        // 队列声明
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定义消费者，实现消费方法
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 当接收到消息后此方法被调用
            
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 交换机
                String exchange = envelope.getExchange();
                // 消息id，mq在channel中标识消费的id
                long deliveryTag = envelope.getDeliveryTag();
                String msg = new String(body, "utf-8");
                System.out.println("consumer: " + msg);
            }
        };
        
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

4.1.3 简单队列不足

耦合性高，生产者和消费者一一对应，无法实现多个消费者消费队列中消息，如果队列名变更，则需要同时都变更

4.2 工作队列 Work queues 4.2.1 模型

多个消费者共同监听一个队列的消息，但一个消息只能被一个消费者消费，不能被重复消费

4.2.2 轮询分发（Round Robin）

生产者

public class Send {
    private static final String QUEUE_NAME = "work_queue_test";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        for (int i = 0; i<50; i++){
            String msg = "hello " + i;
            System.out.println("send-->" + msg);
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            Thread.sleep(100);
        }
        channel.close();
        conn.close();
    }
}

消费者1

public class Receive_1 {
    private static final String QUEUE_NAME = "work_queue_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        DefaultConsumer consumer = 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("[1]receive-->" + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[1] done");
                }
            }
        };
        boolean autoAck = true; // 自动应答
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

消费者2

public class Receive_2 {
    private static final String QUEUE_NAME = "work_queue_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        DefaultConsumer consumer = 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("[2]receive-->" + msg);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[2] done");
                }
            }
        };
        boolean autoAck = true;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

上述先启动消费者1、2，再启动生产者

此时，消息会平均分发给两个消费者，不管消费者消费时间的长短。

4.2.3 公平分发–Fair dispatch

使用公平分发，需要关闭自动应答

生产者

public class Send {
    private static final String QUEUE_NAME = "work_queue_test";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        
        int prefetchCount = 1;
        channel.basicQos(prefetchCount);

        for (int i = 0; i<50; i++){
            String msg = "hello " + i;
            System.out.println("send-->" + msg);
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            Thread.sleep(100);
        }
        channel.close();
        conn.close();
    }
}

消费者1

public class Receive_1 {
    private static final String QUEUE_NAME = "work_queue_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        final Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 保证一次只分发一个
        channel.basicQos(1);

        DefaultConsumer consumer = 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("[1]receive-->" + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[1] done");
                    // 手动确认接收到消息
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

消费者2

public class Receive_2 {
    private static final String QUEUE_NAME = "work_queue_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        final Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 保证一次只分发一个
        channel.basicQos(1);

        DefaultConsumer consumer = 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("[2]receive-->" + msg);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[2] done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

先启动消费者，再启动生产者

此时，当某个消费者将消息消费完之后，才会将消息再分发给该消费者。

4.3 消息应答与消息持久化 4.3.1 消息应答

如上述使用的代码：

boolean autoAck = false;
channel.basicConsume(QUEUE_NAME, autoAck, consumer);

boolean autoAck = true; (自动确认模式)一旦rabbitmq将消息分发给消费者，就会从内存中删除，

这种情况下，当生产者发送完消息，并且杀死正在执行的消费者，就会丢失正在处理的消息。

boolean autoAck = false;(手动确认)，此时消费者手动应答rabbitmq之后，rabbitmq才会删除内存中的这个消息。

4.3.2 消息持久化

boolean durable = false;
channel.queueDeclare(QUEUE_NAME, durable, false, false, null);

注：上述代码中，boolean durable = false;不能直接改为true，改为true，无法运行成功，因为QUEUE_NAME = “work_queue_test”在rabbitmq中已存在，并且是未持久化，rabbitmq不准许重新定义（不同参数，即durable变成了true）一个已存在的队列。

可以重新定义一个队列名称，或者将rabbitmq中已存在的该队列删除。

4.4 订阅模式publish/subscribe 4.4.1 模型

上图中x表示交换机(转发器)。

一个生产者，多个消费者
每一个消费者都有自己的队列
生产者没有直接将消息发送到队列，而是发到交换机exchange
每个队列都要绑定到交换机上
生产者发送的消息，经过交换机，到达队列，就能实现一个消息被多个消费者消费

4.4.2 实现

生产者

public class Send {
    // 交换机名称
    private static final String EXCHANGE_NAME = "exchange_fanout_test";
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        // 声明交换机
        // 第一个参数：交换机名称，
        // 第二个参数：分发类型、fanout表示不处理路由键
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        String msg = "hello world!!!";
        channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());
        System.out.println("send-->" + msg);
        channel.close();
        conn.close();
    }
}

注：此时因为交换机没有绑定队列，并且交换机没有存储能力，所以数据丢失了，rabbitmq中只有队列有存储能力，交换机绑定队列在消费者中完成。

消费者–多个消费者代码都一样，需要注意队列名可能不一样

public class Receive_1 {
    // 队列名
    private static final String QUEUE_NAME = "fanout_queue_test";
    // 交换机名称
    private static final String EXCHANGE_NAME = "exchange_fanout_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        final Channel channel = conn.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 保证一次只分发一个
        channel.basicQos(1);

        DefaultConsumer consumer = 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("[1]receive-->" + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[1] done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

注：先启动消费者，再启动生产者，启动消费者时，rabbitmq中需要存在交换机，否则会报错，无法运行，可以先运行一下生产者，在rabbitmq中创建一个交换机。

4.5 Exchange–交换机

交换机：一方面接收生产者的消息，一方面是向队列推送消息。

channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

fanout：不处理路由键
direct：处理路由键

使用路由模式时需要修改为：channel.exchangeDeclare(EXCHANGE_NAME, "direct");

4.6 路由模式–Routing 4.6.1 模型

4.6.2 实现

生产者

public class Send {
    // 交换机名称
    private static final String EXCHANGE_NAME = "exchange_direct_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        // 声明交换机
        // 第一个参数：交换机名称，第二个参数：分发类型
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");

        String msg = "hello world!!!";

        // 根据设置routingKey，向含有对应routingKey的队列发送消息
        String routingKey = "info";
        channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());
        System.out.println("send-->" + msg);
        channel.close();
        conn.close();
    }
}

消费者，多个消费者代码类似

public class Receive_1 {
    // 队列名
    private static final String QUEUE_NAME = "direct_queue_test";
    // 交换机名称
    private static final String EXCHANGE_NAME = "exchange_direct_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        final Channel channel = conn.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 绑定队列到交换机，可以多次使用绑定多个路由键（routingKey）
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "warning");

        // 保证一次只分发一个
        channel.basicQos(1);

        DefaultConsumer consumer = 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("[1]receive-->" + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[1] done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

4.7 主题模式–Topic exchange

将路由键与某模式匹配

上图中**#表示匹配一个或多个，*** 表示匹配一个

4.7.1 模型

使用这种模式时：channel.exchangeDeclare(EXCHANGE_NAME, "topic");

4.7.2 实现

生产者

public class Send {
    // 交换机名称
    private static final String EXCHANGE_NAME = "exchange_topic_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        // 声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        String msg = "hello world!!!";

        channel.basicPublish(EXCHANGE_NAME, "hello.world", null, msg.getBytes());
        System.out.println("send-->" + msg);
        channel.close();
        conn.close();
    }
}

消费者，多个消费者代码类似

public class Receive_1 {
    // 队列名
    private static final String QUEUE_NAME = "topic_queue_test";
    // 交换机名称
    private static final String EXCHANGE_NAME = "exchange_topic_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        final Channel channel = conn.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "hello.china");
        // 使用 # 匹配一个或多个
        // channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "hello.#");

        // 保证一次只分发一个
        channel.basicQos(1);

        DefaultConsumer consumer = 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("[1]receive-->" + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println("[1] done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

4.8 Header模式

header模式与routing不同的地方在于，header模式取消routingkey，使用header中的 key/value（键值对）匹配队列。

案例：

根据用户的通知设置去通知用户，设置接收Email的用户只接收Email，设置接收sms的用户只接收sms，设置两种通知类型都接收的则两种通知都有效。

代码：

1）生产者

队列与交换机绑定的代码与之前不同，如下：

Map headers_email = new Hashtable();
headers_email.put("inform_type", "email");
Map headers_sms = new Hashtable();
headers_sms.put("inform_type", "sms");
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_HEADERS_INFORM,"",headers_sms);

通知：

String message = "email inform to user"+i;
Map headers =  new Hashtable();
headers.put("inform_type", "email");//匹配email通知消费者绑定的header
//headers.put("inform_type", "sms");//匹配sms通知消费者绑定的header
AMQP.BasicProperties.Builder properties = new AMQP.BasicProperties.Builder();
properties.headers(headers);
//Email通知
channel.basicPublish(EXCHANGE_HEADERS_INFORM, "", properties.build(), message.getBytes());

2）发送邮件消费者

channel.exchangeDeclare(EXCHANGE_HEADERS_INFORM, BuiltinExchangeType.HEADERS);
Map headers_email = new Hashtable();
headers_email.put("inform_email", "email");
//交换机和队列绑定
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_HEADERS_INFORM,"",headers_email);
//指定消费队列
channel.basicConsume(QUEUE_INFORM_EMAIL, true, consumer);

4.8 RPC模式

RPC即客户端远程调用服务端的方法，使用MQ可以实现RPC的异步调用，基于Direct交换机实现，流程如下：

1、客户端即是生产者就是消费者，向RPC请求队列发送RPC调用消息，同时监听RPC响应队列。

2、服务端监听RPC请求队列的消息，收到消息后执行服务端的方法，得到方法返回的结果

3、服务端将RPC方法的结果发送到RPC响应队列

4、客户端（RPC调用方）监听RPC响应队列，接收到RPC调用结果。

五、Rabbitmq的消息确认机制（事务+/confirm/i）

问题：生产者将消息发送出去之后，并不知道消息到底有没有到达rabbitmq。

解决方式：

AMQP实现了事务机制
/confirm/i模式

5.1 事务机制

txSelect：用于将当前channel设置成transaction模式
txCommit：用于提交事务
txRollback：用于回滚事务

缺点：此种方式比较耗时，降低了 rabbitmq 的吞吐量。

实现：

生产者

public class TxSend {
    private static final String QUEUE_NAME = "queue_tx_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        String msg = "hello transaction";

        try {
            channel.txSelect();
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            // 模拟异常
            int i = 1/0;
            System.out.println("send-->" + msg);
            channel.txCommit();
        } catch (Exception ex) {
            channel.txRollback();
            System.out.println("send message rollback");
        }

        channel.close();
        conn.close();
    }
}

消费者

public class TxReceive {
    private static final String QUEUE_NAME = "queue_tx_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("receive-->" + new String(body, "utf-8"));
            }
        };

        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

5.2 /confirm/i模式

普通模式、批量模式、异步模式

/confirm/i是异步的

生产者实现原理：

生产者将信道设置成/confirm/i模式，一旦信道进入/confirm/i模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理；

/confirm/i模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息；

/confirm/i普通模式

// 普通模式，发送一条消息
// 每发送一条消息，调用waitForConfirms()方法等待服务端/confirm/i，这实际上是一种串行的/confirm/i，
// 每publish一条消息之后就等待服务端/confirm/i，如果服务端返回false或者超时时间内未返回，客户端进行消息重传；
public class Send {
    private static final String QUEUE_NAME = "queue_/confirm/i_test";

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 生产者调用/confirm/iSelect将channel设置为/confirm/i模式
        channel./confirm/iSelect();

        String msg = "hello /confirm/i";

        channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
		// 确认
        if(!channel.waitFor/confirm/is()){
            System.out.println("failed");
        } else {
            System.out.println("success");
        }

        channel.close();
        conn.close();
    }
}

/confirm/i批量模式

将普通模式中的channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());用for循环实现

消费者

public class Receive {
    private static final String QUEUE_NAME = "queue_/confirm/i_test";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection conn = ConnectionUtils.getConn();
        Channel channel = conn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("receive-->" + new String(body, "utf-8"));
            }
        };

        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

5.3 异步模式

Channel 对象提供的/confirm/iListener（）回调方法只包含deliveryTag（当前Chanel发出的消息序号），

我们需要自己为每一个Channel维护一个un/confirm/i的消息序号集合，每publish一条数据，集合中元素加1，

每回调一次handleAck方法，un/confirm/i集合删除相应的一条（multiple=false）或多条（multiple=true）记录，

从程序运行效率上看，这个un/confirm/i集合最好采用有序集合SortedSet存储结构

生产者

public class /confirm/iAsySend {
 
	public static final String QUEUE_NAME = "asy_test_/confirm/i";
 
	public static void main(String[] args) throws IOException, TimeoutException {
 
		// 获取连接
		Connection connection = ConnectHelp.getConnect();
 
		// 建立通道
		Channel channel = connection.createChannel();
 
		// 声明队列
		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 
		// 开启事务
		channel./confirm/iSelect();
 
		// 创建set
		final SortedSet confirmSet = Collections.synchronizedSortedSet(new TreeSet());
 
		// 监听事务
		channel.add/confirm/iListener(new /confirm/iListener() {
 
			// handleNack
			public void handleNack(long arg0, boolean arg1) throws IOException {
				// TODO Auto-generated method stub
 
				if (arg1) {
					System.out.println("handleNack : " + arg1);
					/confirm/iSet.headSet(arg0 + 1).clear();
				} else {
					System.out.println("handleNack : " + arg1);
					/confirm/iSet.remove(arg0);
				}
			}
 
			// 没有问题
			public void handleAck(long arg0, boolean arg1) throws IOException {
				// TODO Auto-generated method stub
 
				if (arg1) {
					System.out.println("handleAck :  " + arg1);
					/confirm/iSet.headSet(arg0 + 1).clear();
				} else {
					System.out.println("handleAck :  " + arg1);
					/confirm/iSet.remove(arg0);
				}
 
			}
		});
 
		// 发送的数据
		String msg = "msg data";
 
		while (true) {
			long seqNo = channel.getNextPublishSeqNo();
			channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
			/confirm/iSet.add(seqNo);
		}
 
	}
}

消费者和前面的一样

RabbitMQ使用详解

大数据系统相关栏目本月热门文章