rabbitmq 学习笔记

1。消息会处于阻塞状态，可以通过(man rabbitmqctl 可以获得更多使用方法，常用的有list_queues,list_consumers.list_connections,close_connection,add_vhost,...)

list_channels pid connection ; close_connection [connection] "" ,查看和解决阻塞 

rabbitmqctl list_queues -p [vhost] name messages_ready messages_unacknowledged

解决阻塞的办法，可以在subscribe消息队列是设置autoAck=true，这样会避免消息队列中消息阻塞，这种情况是worker接到消息后，就会把消息从消息队列删除，不管消息是否被正确处理，另一种是设置autoAck=false，这样worker在接受消息后，必须给予服务端一个ack响应，该消息才会从消息队列中删除，这样会防止消息的意外丢失，但要注意的是，消息队列如果没有接收到ack响应，该消息对了的消息就会一直阻塞，对于rabbitmq-server来说，他是没有超时存在的，即除非重启rabbitmq，否则该消息队列会一直阻塞，直到收到响应，但如果与该消息队列的subscirbe断开的话，则表明过期，即该消息队列中消息会尝试重新发消息给一个订阅者进行处理。

2。关于健壮的消息处理
当rabbitmq server重启，或意外当掉的话，所用消息的订阅都会跟着坏掉（当然也可以设置持久化的消息队列设置），解决办法是捕获ShutdownSignalException异常（对rabbitmq）出现该异常说明消息服务无法连到，故可以进行相应的处理，另每次消息发送，消息订阅之前都要进行一次消息队列，exchange，绑定的重定义，防止消息对了重启后改消息队列/exchang已消失。

组合命令

list_consumers -p [vhost]
list_channels -p [vhost] pid connection
list_connections 
close_connection

3。关于消息处理的管理
每个worker都应该有一个可标识的tag，尽量不使用系统生成的，这样便于以后的debug

Connection conn = null;
Channel channel = null;
QueueingConsumer consumer = null;
try {
    conn = qFactory.newConnection();
    channel = conn.createChannel();
    channel.queueDeclare(queueName, false, false, false, null);

    /**
    * This tells RabbitMQ not to give more than one message to a
    * worker at a time. Or, in other words, don't dispatch a new
     * message to a worker until it has processed and acknowledged
    * the previous one. Instead, it will dispatch it to the next
    * worker that is not still busy
    */
        channel.basicQos(1);

    consumer = new QueueingConsumer(channel);
    //Use hostname as consume tagname , So that We can monitor who consume this Queue
                 channel.basicConsume(this.queueName, false, hostName, consumer);
    } catch (IOException e) {
        e.printStackTrace();
    }

    while (true) {
        try {
            //Get next message
            delivery = consumer.nextDelivery();
        } catch (ShutdownSignalException e) {
        //If rabbitmq-server has closed , out of loop
            e.printStackTrace();
            isSignalBroken = true ;
            break;
        } catch (ConsumerCancelledException e) {
            e.printStackTrace();
            log.warn("The consumer has cancelled , Try to re-consume");
            //If the channel and conn have closed .
            try{
            //Sleep 1s and reconnect to rabbitmq-server
                Thread.sleep(1000);
                conn = qFactory.newConnection();
                channel = conn.createChannel();
                            channel.queueDeclare(queueName, false, false, false, null);
                            
                channel.basicQos(1);
                            
                consumer = new  QueueingConsumer(channel);
                        channel.basicConsume(this.queueName, false, consumer);
                continue;
            }catch (IOException e1) {
                e1.printStackTrace();
            } catch (InterruptedException e2) {
                e2.printStackTrace();
            }
                        
        } catch (InterruptedException e) {
            e.printStackTrace();
        }  
                    
                    
        try{
                        
            //process message

        }catch (Exception e) {
            //If throw exception when process message , close channel and conn , make sure this message not block . then re-work
            e.printStackTrace();
            try {
                            channel.basicCancel(hostName);
                channel.close();
                conn.close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
            continue;
        }
                    
    /**
    * If a consumer dies without sending an ack, RabbitMQ will
    * understand that a message wasn't processed fully and will
    * redeliver it to another consumer 
    * There aren't any message timeouts; 
    * RabbitMQ will redeliver the message only when
    * the worker connection dies. It's fine even if processing
    * a message takes a very, very long time
    */
        try {
                         channel.basicAck(delivery.getEnvelope()
                    .getDeliveryTag(), false);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    //If because of the rabbitmq-server stop ,We will re-try connect to rabbtimq-server after 60s
    if(isSignalBroken){
        log.warn("The rabbitmq Server have broken , We Try to re-connect again After 60 seconds");
        try {
            Thread.sleep(1000*60);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        this.run();
    }
}