当前位置: 首页 > news >正文

解决 Python RabbitMQ/Pika 报错:pop from an empty deque

使用 python 的 pika 包连接rabbitmq,代码如下:

import pika
import threading
import timedef on_message(channel, method_frame, header_frame, body):print(f'on_message thread id: {threading.get_ident()}')delivery_tag = method_frame.delivery_tagprint(body, "start")for i in range(10):print(i)time.sleep(20)print(body, "end")channel.basic_ack(delivery_tag)credentials = pika.PlainCredentials('username', 'password')
parameters = pika.ConnectionParameters('test.webapi.username.com', credentials=credentials, heartbeat=5)
connection = pika.BlockingConnection(parameters)channel = connection.channel()
channel.queue_declare(queue="standard", durable=True)
channel.basic_qos(prefetch_count=1)
channel.basic_consume('standard', on_message)print(f'main thread id: {threading.get_ident()}')
try:channel.start_consuming()
except KeyboardInterrupt:channel.stop_consuming()
connection.close()

执行结果:

(Test) [user@user test]$ python mq1.py 
main thread id: 140682766104384
on_message thread id: 140682766104384
b'1' start
0
1
2
3
4
5
6
7
8
9
b'1' end
Traceback (most recent call last):File "mq1.py", line 38, in <module>channel.start_consuming()File "/home/user/conda/envs/Test/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 1865, in start_consumingself._process_data_events(time_limit=None)File "/home/user/conda/envs/Test/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 2026, in _process_data_eventsself.connection.process_data_events(time_limit=time_limit)File "/home/user/conda/envs/Test/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 833, in process_data_eventsself._dispatch_channel_events()File "/home/user/conda/envs/Test/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 567, in _dispatch_channel_eventsimpl_channel._get_cookie()._dispatch_events()File "/home/user/conda/envs/Test/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 1492, in _dispatch_eventsconsumer_info.on_message_callback(self, evt.method,File "mq1.py", line 24, in on_messagechannel.basic_ack(delivery_tag)File "/home/user/conda/envs/Test/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 2112, in basic_ackself._flush_output()File "/home/user/conda/envs/Test/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 1335, in _flush_outputself._connection._flush_output(lambda: self.is_closed, *waiters)File "/home/user/conda/envs/Test/lib/python3.8/site-packages/pika/adapters/blocking_connection.py", line 523, in _flush_outputraise self._closed_result.value.error
pika.exceptions.StreamLostError: Stream connection lost: ConnectionResetError(104, 'Connection reset by peer')

 从结果来看,异常发生在一次长时间的消费过程(200s)完成后报错,具体为调用channel.basic_ack(delivery_tag)发生报错;推测是此时与MQ Server的连接已经被重置ConnectionResetError(104, 'Connection reset by peer'),此时再主动确认就发生报错。

正确解决方案如下:

"""
@author: Zhigang Jiang
@date: 2022/1/16
@description:
"""
import functools
import pika
import threading
import timedef ack_message(channel, delivery_tag):print(f'ack_message thread id: {threading.get_ident()}')if channel.is_open:channel.basic_ack(delivery_tag)else:# Channel is already closed, so we can't ACK this message;# log and/or do something that makes sense for your app in this case.passdef do_work(channel, delivery_tag, body):print(f'do_work thread id: {threading.get_ident()}')print(body, "start")for i in range(10):print(i)time.sleep(20)print(body, "end")cb = functools.partial(ack_message, channel, delivery_tag)channel.connection.add_callback_threadsafe(cb)def on_message(channel, method_frame, header_frame, body):print(f'on_message thread id: {threading.get_ident()}')delivery_tag = method_frame.delivery_tagt = threading.Thread(target=do_work, args=(channel, delivery_tag, body))t.start()credentials = pika.PlainCredentials('username', 'password')
parameters = pika.ConnectionParameters('test.webapi.username.com', credentials=credentials, heartbeat=5)
connection = pika.BlockingConnection(parameters)
channel = connection.channel()
channel.queue_declare(queue="standard", durable=True)
channel.basic_qos(prefetch_count=1)
channel.basic_consume('standard', on_message)print(f'main thread id: {threading.get_ident()}')
try:channel.start_consuming()
except KeyboardInterrupt:channel.stop_consuming()
connection.close()

思路是pika是线程不安全的,所以在接收消息和ACK响应消息时需要另外线程。

相关文章:

解决 Python RabbitMQ/Pika 报错:pop from an empty deque

使用 python 的 pika 包连接rabbitmq&#xff0c;代码如下&#xff1a; import pika import threading import timedef on_message(channel, method_frame, header_frame, body):print(fon_message thread id: {threading.get_ident()})delivery_tag method_frame.delivery_t…...

观察者模式实战

场景 假设创建订单后需要发短信、发邮件等其它的操作&#xff0c;放在业务逻辑会使代码非常臃肿&#xff0c;可以使用观察者模式优化代码 代码实现 自定义一个事件 发送邮件 发送短信 最后再创建订单的业务逻辑进行监听&#xff0c;创建订单 假设后面还需要做其它的…...

035_小驰私房菜_Qualcomm账号注册以及提case流程

全网最具价值的Android Camera开发学习系列资料~ 作者:8年Android Camera开发,从Camera app一直做到Hal和驱动~ 欢迎订阅,相信能扩展你的知识面,提升个人能力~ 一、账号注册 1)登陆高通网站Wireless Technology & Innovation | Mobile Technology | Qualcomm, 采用…...

uniapp input输入框placeholder文本右对齐

input输入框placeholder文本右对齐 给input标签加上placeholder-class&#xff0c;这个是给placeholder设置样式&#xff0c;右对齐这就是text-align:right;字体颜色之类依次编辑即可。...

分布式监控平台—zabbix

前言一、zabbix概述1.1 什么是zabbix1.2 zabbix的监控原理1.3 zabbix常见五个应用程序1.4 zabbix的监控模式1.5 监控架构1.5.1 C/S&#xff08;server—client&#xff09;1.5.2 server—proxy—client1.5.3 master—node—client 二、部署zabbix2.1 部署 zabbix server 端2.2 …...

【leetcode】第一章数组-2

977. 有序数组的平方 简单的方法是平方后使用排序方法第2种方法是双指针方法&#xff0c;从两边进行判断&#xff0c;将最大的从后往前放 public static int[] sortedSquares(int[] nums) {// 输入&#xff1a;nums [-4,-1,0,3,10]// 输出&#xff1a;[0,1,9,16,100]// 解释…...

程序使用Microsoft.XMLHTTP对象请求https时出错解决

程序中使用Microsoft.XMLHTTP组件请求https时出现如下错误&#xff1a; 出错程序代码示例&#xff1a; strUrl "https://www.xxx.com/xxx.asp?id11" dim objXmlHttp set objXmlHttp Server.CreateObject("Microsoft.XMLHTTP") objXmlHttp.open "…...

Linux安装配置nginx+php搭建

Linux安装配置nginxphp搭建 文章目录 Linux安装配置nginxphp搭建1.nginx源码包编译环境和安装相应的依赖1.1 安装编译环境1.2 安装pcre库、zlib库和openssl库 2.安装nginx2.1 在[nginx官网](https://nginx.org/en/download.html)上获取源码包并进行下载2.2 进行解压编译 3.启动…...

springboot的各种配置

1.AOP配置 <!-- AOP的依赖 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-aop</artifactId></dependency>package com.qf.HomeWork.aop;import lombok.extern.slf4j.Slf4j; im…...

OSI七层模型及TCP/IP四层模型

目录 OSI七层模型 TCP/IP四层模型 OIS七层模型和TCP/IP模型图 七层详解 两种模型比较 为什么OSI七层体系结构不常用 四层详解 网络为什么要分层&#xff1f; 说说 OSI 七层模型和 TCP/IP 四层模型的关系和区别 OSI七层模型 OSI&#xff08;Open System Interconnect&a…...

MDN-Web APIs

参考资料 文章目录 简介DOM APIXMLHttpRequestWeb Storage APIWebsockets API 简介 Web APIs&#xff08;Application Programming Interfaces&#xff09;是用于与浏览器环境中的 Web 功能进行交互的一组接口和方法集合。通过 Web APIs&#xff0c;开发人员可以访问浏览器提…...

2023国赛数学建模C题思路分析

文章目录 0 赛题思路1 竞赛信息2 竞赛时间3 建模常见问题类型3.1 分类问题3.2 优化问题3.3 预测问题3.4 评价问题 4 建模资料 0 赛题思路 &#xff08;赛题出来以后第一时间在CSDN分享&#xff09; https://blog.csdn.net/dc_sinor?typeblog 1 竞赛信息 全国大学生数学建模…...

暑假集训笔记

刷题刷的好累啊...不想刷题了...然后就来写题解了... 昨天晚上打了场div2..2000来名&#xff0c;加了155分....现在rating1281...我是菜鸡..暑假之前就打到了1200分以上了&#xff0c;结果暑假一掉再掉&#xff0c;直接掉到1100了...然后我就一直压力很大....... 昨天在机房打…...

【枚举+推式子】牛客小白月赛 63 E

登录—专业IT笔试面试备考平台_牛客网 题意&#xff1a; 思路&#xff1a; 首先是个计数问题&#xff0c;考虑组合数学 组合数学就是在考虑枚举所有包含1和n的区间 这个典中典就是枚举1和n的位置然后算贡献 双指针超时&#xff0c;考虑推式子&#xff1a; Code&#xff1a…...

Android多屏幕支持-Android12

Android多屏幕支持-Android12 1、概览及相关文章2、屏幕窗口配置2.1 配置xml文件2.2 DisplayInfo#uniqueId 屏幕标识2.3 adb查看信息 3、配置文件解析3.1 xml字段读取3.2 简要时序图 4、每屏幕焦点 android12-release 1、概览及相关文章 AOSP > 文档 > 心主题 > 多屏…...

python环境下载安装教程,python运行环境怎么下载

本篇文章给大家谈谈python安装步骤以及环境变量配置&#xff0c;以及下载python需要设置环境变量吗&#xff0c;希望对各位有所帮助&#xff0c;不要忘了收藏本站喔。 1.https://www.python.org/downloads/windows/ 下载适合自己电脑的python安装包 2.下载后安装即可 3.配置环…...

【0.2】lubancat鲁班猫4远程ubuntu22.04.2 无需任何安装

环境 lubancat4鲁班猫4 (4G0)不带emmc 系统镜像ubuntu-22.04.2-desktop-arm64-lubancat-4.img 网络环境:有线网络与本win10电脑同意环境 操作步骤ubuntu正常开机登陆用户&#xff0c;连接好网络进入设置>网络查看设备当前局域网IP 如192.168.199.159进入设置>共享>远程…...

Flutter 状态管理 Provider

状态管理必要性 Flutter基于声明式构建UI&#xff0c;原生则是命令式&#xff0c;状态管理是用于解决声明式开发带来的问题。 例&#xff1a;命令式的原生&#xff0c;数据更新需要拿到对应控件并更改其显示值&#xff1b;而声明式则需要更改数据值并通过setstate更新状态&am…...

【设计模式】观察者模式

什么是观察者模式&#xff1f; 观察者模式&#xff08;又被称为发布-订阅&#xff08;Publish/Subscribe&#xff09;模式&#xff0c;属于行为型模式的一种&#xff0c;它定义了一种一对多的依赖关系&#xff0c;让多个观察者对象同时监听某一个主题对象。这个主题对象在状态…...

ORCA优化器浅析——CDXLOperator Base class for operators in a DXL tree

如上图所示&#xff0c;CDXLOperator作为Base class for operators in a DXL tree&#xff0c;其子类CDXLLogical、CDXLScalar、CDXLPhysical作为逻辑节点、物理节点和Scalar节点的DXL表示类&#xff0c;因此其包含了这些类的共同部分特性&#xff0c;比如获取其DXL节点表示的函…...

Lombok 的 @Data 注解失效,未生成 getter/setter 方法引发的HTTP 406 错误

HTTP 状态码 406 (Not Acceptable) 和 500 (Internal Server Error) 是两类完全不同的错误&#xff0c;它们的含义、原因和解决方法都有显著区别。以下是详细对比&#xff1a; 1. HTTP 406 (Not Acceptable) 含义&#xff1a; 客户端请求的内容类型与服务器支持的内容类型不匹…...

日语学习-日语知识点小记-构建基础-JLPT-N4阶段(33):にする

日语学习-日语知识点小记-构建基础-JLPT-N4阶段(33):にする 1、前言(1)情况说明(2)工程师的信仰2、知识点(1) にする1,接续:名词+にする2,接续:疑问词+にする3,(A)は(B)にする。(2)復習:(1)复习句子(2)ために & ように(3)そう(4)にする3、…...

Java多线程实现之Callable接口深度解析

Java多线程实现之Callable接口深度解析 一、Callable接口概述1.1 接口定义1.2 与Runnable接口的对比1.3 Future接口与FutureTask类 二、Callable接口的基本使用方法2.1 传统方式实现Callable接口2.2 使用Lambda表达式简化Callable实现2.3 使用FutureTask类执行Callable任务 三、…...

前端开发面试题总结-JavaScript篇(一)

文章目录 JavaScript高频问答一、作用域与闭包1.什么是闭包&#xff08;Closure&#xff09;&#xff1f;闭包有什么应用场景和潜在问题&#xff1f;2.解释 JavaScript 的作用域链&#xff08;Scope Chain&#xff09; 二、原型与继承3.原型链是什么&#xff1f;如何实现继承&a…...

AI编程--插件对比分析:CodeRider、GitHub Copilot及其他

AI编程插件对比分析&#xff1a;CodeRider、GitHub Copilot及其他 随着人工智能技术的快速发展&#xff0c;AI编程插件已成为提升开发者生产力的重要工具。CodeRider和GitHub Copilot作为市场上的领先者&#xff0c;分别以其独特的特性和生态系统吸引了大量开发者。本文将从功…...

AspectJ 在 Android 中的完整使用指南

一、环境配置&#xff08;Gradle 7.0 适配&#xff09; 1. 项目级 build.gradle // 注意&#xff1a;沪江插件已停更&#xff0c;推荐官方兼容方案 buildscript {dependencies {classpath org.aspectj:aspectjtools:1.9.9.1 // AspectJ 工具} } 2. 模块级 build.gradle plu…...

今日学习:Spring线程池|并发修改异常|链路丢失|登录续期|VIP过期策略|数值类缓存

文章目录 优雅版线程池ThreadPoolTaskExecutor和ThreadPoolTaskExecutor的装饰器并发修改异常并发修改异常简介实现机制设计原因及意义 使用线程池造成的链路丢失问题线程池导致的链路丢失问题发生原因 常见解决方法更好的解决方法设计精妙之处 登录续期登录续期常见实现方式特…...

【数据分析】R版IntelliGenes用于生物标志物发现的可解释机器学习

禁止商业或二改转载&#xff0c;仅供自学使用&#xff0c;侵权必究&#xff0c;如需截取部分内容请后台联系作者! 文章目录 介绍流程步骤1. 输入数据2. 特征选择3. 模型训练4. I-Genes 评分计算5. 输出结果 IntelliGenesR 安装包1. 特征选择2. 模型训练和评估3. I-Genes 评分计…...

第7篇:中间件全链路监控与 SQL 性能分析实践

7.1 章节导读 在构建数据库中间件的过程中&#xff0c;可观测性 和 性能分析 是保障系统稳定性与可维护性的核心能力。 特别是在复杂分布式场景中&#xff0c;必须做到&#xff1a; &#x1f50d; 追踪每一条 SQL 的生命周期&#xff08;从入口到数据库执行&#xff09;&#…...

通过MicroSip配置自己的freeswitch服务器进行调试记录

之前用docker安装的freeswitch的&#xff0c;启动是正常的&#xff0c; 但用下面的Microsip连接不上 主要原因有可能一下几个 1、通过下面命令可以看 [rootlocalhost default]# docker exec -it freeswitch fs_cli -x "sofia status profile internal"Name …...