Spring整合RabbitMQ 一直收到重复消息怎么处理？ - 开源中国社区
当前访客身份：游客 [
当前位置：
使用Spring的AqmpTemplate整合rabbitMQ，然后指定了一个listener，但是现在一直收到重复的消息，请问怎么破？
&!-- rabbit MQ --&
&!-- 连接服务配置 --&
&rabbit:connection-factory id="connectionFactory"
host="${mq.host}" username="${mq.username}" password="${mq.password}"
port="${mq.port}" /&
&rabbit:admin connection-factory="connectionFactory" /&
&!-- queue 队列声明 --&
&rabbit:queue id="spr_queue" durable="true" auto-delete="false"
exclusive="false" name="spr_queue" /&
&!-- exchange queue binging key 绑定 --&
&rabbit:direct-exchange name="spr_exchange"
durable="true" auto-delete="false" id="spr_exchange"&
&rabbit:bindings&
&rabbit:binding queue="spr_queue" key="queue_one_key" /&
&/rabbit:bindings&
&/rabbit:direct-exchange&
&rabbit:template exchange="spr_exchange" id="amqpTemplate"
connection-factory="connectionFactory" /&
&rabbit:listener-container
connection-factory="connectionFactory" acknowledge="auto"&
&rabbit:listener queues="spr_queue" ref="queueOneLitener" /&
&/rabbit:listener-container&
&bean id="queueOneLitener" class="com.my.base.RabbitMqListener" /&
二、发送代码
amqpTemplate.convertAndSend("queue_one_key", "Hello queue!");
三、监听器
public class RabbitMqListener implements ChannelAwareMessageListener {
public void onMessage(Message message, Channel channel) throws Exception {
String str = new String(message.getBody());
// TODO 接受消息
System.out.println(message.getMessageProperties().getMessageId());
System.out.println("Recived msg:" + str);
现在一直收到同一条重复消息，控制台一直打印
Recived msg:Hello queue!
Recived msg:Hello queue!
Recived msg:Hello queue!
Recived msg:Hello queue!
Recived msg:Hello queue!
Recived msg:Hello queue!
Recived msg:Hello queue!
Recived msg:Hello queue!
Recived msg:Hello queue!
Recived msg:Hello queue!
Recived msg:Hello queue!
请问是哪里的配置有什么问题吗？我是参考这位大侠的文章进行配置的
http://my.oschina.net/yilian/blog/174244
共有4个答案
<span class="a_vote_num" id="a_vote_num_
<span class="a_vote_num" id="a_vote_num_
同问呐，这些框架的重复消费如何解决
--- 共有 1 条评论 ---
你把ack设为自动的，然后再看下处理消息的时候有没有抛异常
(3个月前)&nbsp&
<span class="a_vote_num" id="a_vote_num_
onMessage最后来一句
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
<span class="a_vote_num" id="a_vote_num_
监听的代码改成这样试试
public class RabbitMqListener implements MessageListener {
public void onMessage(Message message) {
String str = new String(message.getBody());
// TODO 接受消息
System.out.println(message.getMessageProperties().getMessageId());
System.out.println("Recived msg:" + str);
} catch (Exception e) {
// TODO: handle exception
更多开发者职位上
有什么技术问题吗？
_汉唐_的其它问题
类似的话题&&&&spring amqp
rabbit 消息确认及外部事物
&spring amqp
rabbit 消息确认及外部事物
spring rabbitmq 消息确认 外部事物
若举报审核通过，可奖励20下载分
被举报人：
convict_eva
举报的资源分：
请选择类型
资源无法下载
资源无法使用
标题与实际内容不符
含有危害国家安全内容
含有反动色情等内容
含广告内容
版权问题，侵犯个人或公司的版权
*详细原因：
VIP下载&&免积分60元/年（1200次）
您可能还需要
Q.为什么我点的下载下不了，但积分却被扣了
A. 由于下载人数众多，下载服务器做了并发的限制。若发现下载不了，请稍后再试，多次下载是不会重复扣分的。
Q.我的积分不多了，如何获取积分？
A. 获得积分，详细见。
完成任务获取积分。
论坛可用分兑换下载积分。
第一次绑定手机，将获得5个C币，C币可。
关注并绑定CSDNID，送10个下载分
下载资源意味着您已经同意遵守以下协议
资源的所有权益归上传用户所有
未经权益所有人同意，不得将资源中的内容挪作商业或盈利用途
CSDN下载频道仅提供交流平台，并不能对任何下载资源负责
下载资源中如有侵权或不适当内容，
本站不保证本站提供的资源的准确性，安全性和完整性，同时也不承担用户因使用这些下载资源对自己和他人造成任何形式的伤害或损失。
开发技术下载排行
您当前C币：0&&&可兑换 0 下载积分
兑换下载分:&
消耗C币:0&
立即兑换&&
兑换成功你当前的下载分为 。前去下载资源
你下载资源过于频繁，请输入验证码
如何快速获得积分？
你已经下载过该资源，再次下载不需要扣除积分
spring amqp
rabbit 消息确认及外部事物
所需积分：1
剩余积分：0
扫描微信二维码精彩活动、课程更新抢先知
VIP会员，免积分下载
会员到期时间：日
剩余下载次数：1000
spring amqp
rabbit 消息确认及外部事物
剩余次数：&&&&有效期截止到：
你还不是VIP会员VIP会员享免积分 . 专属通道极速下载
VIP下载次数已满VIP会员享免积分 . 专属通道极速下载，请继续开通VIP会员
你的VIP会员已过期VIP会员享免积分 . 专属通道极速下载，请继续开通VIP会员Rabbitmq使用小结
Rabbitmq使用小结
1. 神奇的现象
今天发布版本时，重启了整个rabbitmq集群，发现有的模块和rabbitmq中的exchange解绑了。而且启动时并没有自动绑定。有个同事有十几个queue，全部手动绑定了一下。导致上线时间多花了10多分钟。现象描述：
queue和exchange自动解绑
启动时没有自动绑定queue
2. 背后的故事
rabbitmq我们采用的是fanout exchange模式，这样一个应用发消息，其他模块感兴趣的可以自己创建queue去绑定监听。现在用户产生一条数据，后面的很多服务比如说Feed，搜索，计数，行为记录，系统通知等都采用了异步处理的方式。也说明了我们整个架构依赖于Rabbitmq。
rabbitmq的exchange，queue有durable，非durable两种。我们自动解绑的原因就是因为exchange设置的是非durable的，导致每次重启后，绑定都失效了。但是发送方又自动创建了exchange，而queue是durable的，现象就是她们之间失去了绑定。
用rabbitmq开发时，java端用了spring amqp，ruby端用了bunny。我们就说spring的amqp吧，在rabbitmq使用中有以下几点需要注意：
刚才也说到，有的项目没有自动绑定queue，原因有两个（1）没有配置amqpadmin，（2）lazy-init设置的是true。spring的amqpadmin在系统启动时会自动get出exchange，queue的bean，将她们在rabbitmq中注册。
& & & &&rabbit:admin id="amqpAdmin" connection-factory="connectionFactory" /&
& & &2. Spring在这里也使用了它那万能的template模式，connectionfactory用的是CachingConnectionFactory，cachingconnectionfactory的cachingsize要注意他不是连接数的大小，是channel的个数，channel相当于session的概念，一个connection可以创建多个channel。cachingconnectionfactory用aop的方式用一个双向链表pool了cachesize的channel。我们在它的基础上用commons-pool开发了一个连接池。不过cachingconnectionfactory性能已经够好了。
& & &3. 对于监听方Spring又使用了他那万能的simplelistenercontainer，估计还是轮询不断receive消息。concurrentconsumer可以配多个线程并发收消息，假设是收到消息的处理逻辑时间较长。可以多线程并发的处理，提高处理效率。 &
3. 真实的世界
& &真实使用时，rabbitmq如果出问题了，那我们整个系统的数据就不同步了。甚至一些主要功能都会无法正常试用。所以queue要处理好单点失败的问题。
& & & & &（1）rabbitmq我们采用的是active/active的模式，一些重要的queue可以采用mirror的方式复制。保证高可用。如果不是mirror queue的方式，就有可能会丢失数据。如果用了mirror queue，即使是ram node也会在磁盘里写入数据。
& & & & &（2）接收方负载均衡，接收方我们可以监听集群中任意rabbitmq，写入的服务器会平均的把消息deliver给其他node。两个node测试下来，各处理了50%。其他node其实也相当于一个listener。这个逻辑还需要多测试，比如一台多配一些listener，处理的快一点，试试是否还是绝对平均。
& & & & &（3）我们会发现在配置写入的client端时，我们还是指定的其中的某台rabbitmq服务器用于写入，就像主库一样。但是它有可能会是单点失败的。这时我们往往是在前面加个HAProxy或者LVS来解决问题。这样写入就做了负载均衡，也解决了failover的问题。HAProxy通过keeplive的方式保持高可用。
& & 很多人喜欢用代码来自动创建exchange，queue，binding，我其实不太喜欢这样，我觉得这些就像契约一样，应该预先商定好，创建好。（1）大家配的不是相同的，（2）可能会乱配，要求durable的，有的人配的不是durable，（3）应该是exchange，有的人用在了queue上。还是线下陪比较稳妥。
发表评论：
TA的最新馆藏[转]&}