RabbitMQ笔记二十四：RabbitMQ异常处理

异常处理

RabbitMQ java client中的异常处理

消费消息，在消费消息的时候抛出异常，

消费启动类：

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

import java.util.concurrent.TimeUnit;

public class Consumer {
    public static void main(String[] args) throws Exception{

        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUri("amqp://zhihao.miao:123456@192.168.1.131:5672");

        Connection connection = connectionFactory.newConnection();


        Channel channel = connection.createChannel();

        System.out.println(channel.isOpen());

        channel.basicConsume("sms",true,new SimpleConsumer(channel));

        TimeUnit.SECONDS.sleep(20);
        System.out.println(channel.isOpen());

        channel.close();
        connection.close();

    }
}

消费逻辑

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

import java.io.IOException;

public class SimpleConsumer extends DefaultConsumer{

    public SimpleConsumer(Channel channel){
        super(channel);
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope,
                               AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.out.println(consumerTag);
        System.out.println("-----收到消息了-----------");
        System.out.println("消息属性为："+properties);
        System.out.println("消息内容为："+new String(body));

        throw new NullPointerException("空指针异常");
    }
}

然后我们发现20s过后，发现控制台打印了

true
amq.ctag-fJ45VlTAV2aKO97-zztDNQ
-----收到消息了-----------
消息属性为：#contentHeader<basic>(content-type=null, content-encoding=null, headers={}, delivery-mode=1, priority=null, correlation-id=null, reply-to=null, expiration=null, message-id=null, timestamp=null, type=null, user-id=null, app-id=null, cluster-id=null)
消息内容为：第一条消息
Exception in thread "main" com.rabbitmq.client.AlreadyClosedException: channel is already closed due to clean channel shutdown; protocol method: #method<channel.close>(reply-code=200, reply-text=Closed due to exception from Consumer com.zhihao.test.day10.SimpleConsumer@2462e0fd (amq.ctag-fJ45VlTAV2aKO97-zztDNQ) method handleDelivery for channel AMQChannel(amqp://zhihao.miao@192.168.1.131:5672/,1), class-id=0, method-id=0)
    at com.rabbitmq.client.impl.AMQChannel.processShutdownSignal(AMQChannel.java:286)
false
    at com.rabbitmq.client.impl.ChannelN.startProcessShutdownSignal(ChannelN.java:282)
    at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:596)
    at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:530)
    at com.rabbitmq.client.impl.ChannelN.close(ChannelN.java:523)
    at com.rabbitmq.client.impl.recovery.AutorecoveringChannel.close(AutorecoveringChannel.java:68)
    at com.zhihao.test.day10.Consumer.main(Consumer.java:31)

我们发现当消费端抛出异常的时候，channel会关闭，然后channel.close()会报错。原因是什么呢？

我们知道com.rabbitmq.client.ExceptionHandler这个接口，中定义了各个阶段异常的捕获方法。其默认实现com.rabbitmq.client.impl.DefaultExceptionHandler，继承com.rabbitmq.client.impl.StrictExceptionHandler，发现当消费失败的时候会kill掉channel。

源码如下：

@Override
    public void handleConsumerException(Channel channel, Throwable exception,
                                        Consumer consumer, String consumerTag,
                                        String methodName)
    {
        handleChannelKiller(channel, exception, "Consumer " + consumer
                                                        + " (" + consumerTag + ")"
                                                        + " method " + methodName
                                                        + " for channel " + channel);
    }

调用handleChannelKiller方法

  @Override
    protected void handleChannelKiller(Channel channel, Throwable exception, String what) {
        log(what + " threw an exception for channel " + channel, exception);
        try {
            channel.close(AMQP.REPLY_SUCCESS, "Closed due to exception from " + what);
        } catch (AlreadyClosedException ace) {
            // noop
        } catch (TimeoutException ace) {
            // noop
        } catch (IOException ioe) {
            log("Failure during close of channel " + channel + " after " + exception, ioe);
            channel.getConnection().abort(AMQP.INTERNAL_ERROR, "Internal error closing channel for " + what);
        }
    }

消息发送确认异常捕获：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.impl.DefaultExceptionHandler;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

public class Send {
    public static void main(String[] args) throws Exception{
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUri("amqp://zhihao.miao:123456@192.168.1.131:5672");

        connectionFactory.setExceptionHandler(new DefaultExceptionHandler(){
            @Override
            public void handleConfirmListenerException(Channel channel, Throwable exception) {
                System.out.println("=====消息确认发生异常=======");
                exception.printStackTrace();
            }
        });


        Connection connection = connectionFactory.newConnection();

        Channel channel = connection.createChannel();

        channel.confirmSelect();

        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("收到消息确认，："+deliveryTag);
                throw new IOException("数据库异常，确认失败");
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {

            }
        });

        channel.basicPublish("","sms",null,"发送消息".getBytes());


        TimeUnit.SECONDS.sleep(20);

        channel.close();
        connection.close();


    }
}

Spring AMQP异常处理

设置AUTO确认的时候已经讲解了异常处理，这边主要讲解一下自动声明的时候的异常处理。

正常的情况下我们声明队列的代码：

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class MQConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setUri("amqp://zhihao.miao:123456@192.168.1.131:5672");
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public Queue infoQueue(){
        Map<String,Object> arguments = new HashMap<>();
        return new Queue("info",true,false,false,arguments);
    }

    @Bean
    public Queue errorQueue(){
        Map<String,Object> arguments = new HashMap<>();
        return new Queue("error",true,false,false,arguments);
    }
}

应用启动类：

import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.ComponentScan;

@ComponentScan
public class Application {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class);
        //使得客户端第一次连接rabbitmq
        context.getBean(RabbitAdmin.class).getQueueProperties("**");
        context.close();
    }
}

执行声明成功，但是假如我们的info队列已经存在，并且属性和自动声明的不一致，那么就会抛出异常造成info声明不了，糟糕的是error队列也声明不了。

已经声明了info队列

此时执行发现抛出异常，并且声明不了队列。

控制台打印出异常堆栈信息：
信息: Starting beans in phase -2147482648
九月 29, 2017 11:49:30 下午 org.springframework.amqp.rabbit.connection.CachingConnectionFactory createBareConnection
信息: Created new connection: connectionFactory#163e4e87:0/SimpleConnection@ef9296d [delegate=amqp://zhihao.miao@192.168.1.131:5672/, localPort= 63190]
九月 29, 2017 11:49:30 下午 org.springframework.amqp.rabbit.connection.CachingConnectionFactory log
严重: Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-dead-letter-exchange' for queue 'info' in vhost '/': received none but current is the value 'zhihao.topic.exchange' of type 'longstr', class-id=50, method-id=10)
Exception in thread "main" org.springframework.amqp.AmqpIOException: java.io.IOException
    at org.springframework.amqp.rabbit.support.RabbitExceptionTranslator.convertRabbitAccessException(RabbitExceptionTranslator.java:71)
    at org.springframework.amqp.rabbit.connection.RabbitAccessor.convertRabbitAccessException(RabbitAccessor.java:113)
    at org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:1461)
    at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1411)
    at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1387)
    at org.springframework.amqp.rabbit.core.RabbitAdmin.initialize(RabbitAdmin.java:500)
    at org.springframework.amqp.rabbit.core.RabbitAdmin$11.onCreate(RabbitAdmin.java:419)
    at org.springframework.amqp.rabbit.connection.CompositeConnectionListener.onCreate(CompositeConnectionListener.java:33)
    at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createConnection(CachingConnectionFactory.java:571)
    at org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:1430)
    at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1411)
    at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:1387)
    at org.springframework.amqp.rabbit.core.RabbitAdmin.getQueueProperties(RabbitAdmin.java:336)
    at com.zhihao.miao.exception.Application.main(Application.java:18)

怎样去解决呢？

修改配置类，设置忽略声明异常

import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class MQConfig {

    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory factory = new CachingConnectionFactory();
        factory.setUri("amqp://zhihao.miao:123456@192.168.1.131:5672");
        return factory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        //设置忽略声明异常
        rabbitAdmin.setIgnoreDeclarationExceptions(true);
        return rabbitAdmin;
    }

    @Bean
    public Queue infoQueue(){
        Map<String,Object> arguments = new HashMap<>();
        return new Queue("info",true,false,false,arguments);
    }

    @Bean
    public Queue errorQueue(){
        Map<String,Object> arguments = new HashMap<>();
        return new Queue("error",true,false,false,arguments);
    }
}

重新启动应用启动类，

import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.ComponentScan;

@ComponentScan
public class Application {
    public static void main(String[] args) {
        AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(Application.class);
        //使得客户端第一次连接rabbitmq
        context.getBean(RabbitAdmin.class).getQueueProperties("**");
        context.close();
    }
}

此时发现队列error声明成功，info声明失败，控制台打印：

九月 29, 2017 11:51:23 下午 org.springframework.context.annotation.AnnotationConfigApplicationContext prepareRefresh
信息: Refreshing org.springframework.context.annotation.AnnotationConfigApplicationContext@446cdf90: startup date [Fri Sep 29 23:51:23 CST 2017]; root of context hierarchy
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
九月 29, 2017 11:51:24 下午 org.springframework.context.support.DefaultLifecycleProcessor start
信息: Starting beans in phase -2147482648
九月 29, 2017 11:51:24 下午 org.springframework.amqp.rabbit.connection.CachingConnectionFactory createBareConnection
信息: Created new connection: connectionFactory#163e4e87:0/SimpleConnection@ef9296d [delegate=amqp://zhihao.miao@192.168.1.131:5672/, localPort= 63244]
九月 29, 2017 11:51:24 下午 org.springframework.amqp.rabbit.connection.CachingConnectionFactory log
严重: Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-dead-letter-exchange' for queue 'info' in vhost '/': received none but current is the value 'zhihao.topic.exchange' of type 'longstr', class-id=50, method-id=10)
九月 29, 2017 11:51:24 下午 org.springframework.amqp.rabbit.core.RabbitAdmin logOrRethrowDeclarationException
警告: Failed to declare queue: Queue [name=info, durable=true, autoDelete=false, exclusive=false, arguments={}], continuing... com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'x-dead-letter-exchange' for queue 'info' in vhost '/': received none but current is the value 'zhihao.topic.exchange' of type 'longstr', class-id=50, method-id=10)
九月 29, 2017 11:51:24 下午 org.springframework.context.annotation.AnnotationConfigApplicationContext doClose
信息: Closing org.springframework.context.annotation.AnnotationConfigApplicationContext@446cdf90: startup date [Fri Sep 29 23:51:23 CST 2017]; root of context hierarchy
九月 29, 2017 11:51:24 下午 org.springframework.context.support.DefaultLifecycleProcessor stop
信息: Stopping beans in phase -2147482648

此时也就达到了我们的需求。

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 211,423评论 6赞 491
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,147评论 2赞 385
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 157,019评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,443评论 1赞 283
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,535评论 6赞 385
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,798评论 1赞 290
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,941评论 3赞 407
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,704评论 0赞 266
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,152评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,494评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,629评论 1赞 340
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,295评论 4赞 329
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,901评论 3赞 313
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,742评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,978评论 1赞 266
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,333评论 2赞 360
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,499评论 2赞 348

推荐阅读更多精彩内容

  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 134,629评论 18赞 139
  • RabbitMQ Java客户端API指南
    本文章翻译自http://www.rabbitmq.com/api-guide.html，并没有及时更新。 术语对...
    joyenlee阅读 7,645评论 0赞 3
  • RabbitMQ
    来源 RabbitMQ是用Erlang实现的一个高并发高可靠AMQP消息队列服务器。支持消息的持久化、事务、拥塞控...
    jiangmo阅读 10,346评论 2赞 34
  • 消息队列之 RabbitMQ
    关于消息队列，从前年开始断断续续看了些资料，想写很久了，但一直没腾出空，近来分别碰到几个朋友聊这块的技术选型，是时...
    预流阅读 584,522评论 51赞 785
  • Rabbitmq学习笔记
    什么叫消息队列 消息（Message）是指在应用间传送的数据。消息可以非常简单，比如只包含文本字符串，也可以更复杂...
    lijun_m阅读 1,340评论 0赞 1