正文
python万能消费框架，新增7种中间件(或操作mq的包)和三种并发模式。
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
新增的中间件和并发模式见注释。
消息队列中间件方面celery支持的，都要支持。并发模式，celery支持的都要支持。
从无限重复相似代码抽取框架，做成万能复用，是生产力的保障。
使用模板模式使加新中间件时候，在改实现消费框架的代码非常方便，不会影响到原有中间件使用。
使用策略模式使加入新的并发模式，，在改实现消费框架的代码非常方便，不会影响到原有并发模式。
所以实现消费框架的代码虽然很长有1000多行，但修改和增加的时候不会出现如履薄冰的害怕情绪。
使用工厂模式，使得调用框架时候，非常容易切换基于不同消息中间件的使用，只需要改一个数字就改变消费和推送代码使用的中间件。
使快速测试不同种类的中间件和并发方式变得很容易。
7种中间件包括使用pika rabbitpy aqpstorm操作rabbitmq、基于redis的list数据结构、基于mongo queue包实现的mongo消息队列 、基于python Queue对象的消息队列（随着python解释器退出而消失）、基于使用persitqueue包实现的sqllite3本地持久化队列。
3种并发模式为thread、gevent、evenlet模式。（支持基于多进程的分布式模式，由于启动进程必须是在___name__ = main里面，所以需要用户自己写，自己写Process(target=f).start()。如果要使用多进程，一般也是前面三种模式 加 进程模式配合，例如多进程 + gevent ，只有100%纯cpu计算的才适合纯多进程。）
使用多进程的目的，开32进程，最高可以使双路e5 32核 cpu使用率达到3200%， 将cpu打满 ，充分利用cpu资源。 如果不开多进程，就算程序忙的要命，cpu使用率过不了110%，浪费cpu资源。
# -*- coding: utf-8 -*-
# @Author : ydf """
类celery的worker模式，可用于一切需要分布式并发的地方，最好是io类型的。可以分布式调度起一切函数。
rabbitmq生产者和消费者框架。完全实现了celery worker模式的全部功能，使用更简单。支持自动重试指定次数，
消费确认，指定数量的并发线程，和指定频率控制1秒钟只运行几次， 同时对mongodb类型的异常做了特殊处理
最开始写得是使用pika包，非线程安全,后来加入rabbitpy,rabbitpy包推送会丢失部分数据，推荐pika包使用
单下划线代表保护，双下划线代表私有。只要关注公有方法就可以，其余是类内部自调用方法。 3月15日
1）、新增RedisConsumer 是基于redis中间件的消费框架，不支持随意暂停程序或者断点，会丢失一部分正在运行中的任务，推荐使用rabbitmq的方式。
get_consumer是使用工厂模式来生成基于rabbit和reids的消费者，使用不同中间件的消费框架更灵活一点点，只需要修改一个数字。 3月20日
2）、增加支持函数参数过滤的功能，可以随时放心多次推送相同的任务到中间件，会先检查该任务是否需要执行，避免浪费cpu和流量，加快处理速度。
基于函数参数值的过滤，需要设置 do_task_filtering 参数为True才生效，默认为False。
3）、新增支持了函数的参数是多个参数，需要设置is_consuming_function_use_multi_params 为True生效，为了兼容老代码默认为False。
区别是消费函数原来需要
def f(body): # 函数有且只能有一个参数，是字典的多个键值对来表示参数的值。
print(body['a'])
print(body['b']) 现在可以
def f(a,b):
print(a)
print(b) 对于推送的部分,都是一样的，都是推送 {"a":1,"b":2} 开启消费都是 get_consumer('queue_test', consuming_function=f).start_consuming_message() 6月3日
1) 增加了RedisPublisher类，和增加get_publisher工厂模式
方法同mqpublisher一样，这是为了增强一致性，以后每个业务的推送和消费，
如果不直接使用RedisPublisher RedisConsumerer RabbitmqPublisher RabbitMQConsumer这些类，而是使用get_publisher和get_consumer来获取发布和消费对象，
支持修改一个全局变量的broker_kind数字来切换所有平台消费和推送的中间件种类。
2）增加指定不运行的时间的配置。例如可以白天不运行，只在晚上运行。
3）增加了函数超时的配置，当函数运行时间超过n秒后，自动杀死函数，抛出异常。
4) 增加每分钟函数运行次数统计，和按照最近一分钟运行函数次数来预估多久可以运行完成当前队列剩余的任务。
5） 增加一个判断函数，阻塞判断连续多少分钟队列里面是空的。判断任务疑似完成。
6）增加一个终止消费者的标志，设置标志后终止循环调度消息。
7) consumer对象增加内置一个属性，表示相同队列名的publisher实例。 6月29日
1） 增加消息过期时间的配置，消费时候距离发布时候超过一定时间，丢弃任务。
2）增加基于python内置Queue对象的本地队列作为中间件的发布者和消费者，公有方法的用法与redis和mq的完全一致，
方便没有安装mq和redis的环境使用测试除分布式以外的其他主要功能。使用内置queue无法分布式和不支持程序重启任务接续。
好处是可以改一个数字就把代码运行起来在本地测试，不会接受和推送消息到中间件影响别人，别人也影响不了自己，自测很合适。
3）实例化发布者时候，不在初始化方法中链接中间件，延迟到首次真正使用操作中间件的方法。
4)BoundedThreadpoolExecutor替换成了新的CustomThreadpoolExecutor 7月2日
加入了gevent并发模式，设置concurrent_mode为2生效。 7月3日
加入了evenlet并发模式，设置concurrent_mode为3生效。 7月4日
1）增加使用amqpstorm实现的rabbit操作的中间件，设置broker_kind为4生效，支持消费确认
2）增加mongo-queue实现的mongodb为中间件的队列，设置broker_kind为5生效，支持确认消费
3）增加persistqueue sqllite3实现的本地持久化队列，支持多进程和多次启动不在同一个解释器下的本地分布式。比python内置Queue对象增加了持久化和支持不同启动批次的脚本推送 消费。sqllite不需要安装这个中间件就可以更方便使用。设置broker_kind为6生效，支持确认消费。 """
# import functools
import abc
# import atexit
import atexit
import copy
from queue import Queue
import threading
import gevent
import eventlet
import traceback
import typing
import json
from collections import Callable, OrderedDict
import time
from functools import wraps
from threading import Lock, Thread
import unittest from mongomq import MongoQueue # pip install mongo-mq==0.0.1
import sqlite3
import persistqueue # pip install persist-queue==0.4.2
import amqpstorm # pip install AMQPStorm==2.7.1
from amqpstorm.basic import Basic as AmqpStormBasic
from amqpstorm.queue import Queue as AmqpStormQueue
import rabbitpy
from pika import BasicProperties
# noinspection PyUnresolvedReferences
from pika.exceptions import ChannelClosed
# from rabbitpy.message import Properties
import pika
from pika.adapters.blocking_connection import BlockingChannel
from pymongo.errors import PyMongoError
from app.utils_ydf import (LogManager, LoggerMixin, RedisMixin, RedisBulkWriteHelper, RedisOperation, decorators, time_util, LoggerLevelSetterMixin, nb_print, CustomThreadPoolExecutor, MongoMixin)
# noinspection PyUnresolvedReferences
from app.utils_ydf import BoundedThreadPoolExecutor, block_python_exit
from app.utils_ydf.custom_evenlet_pool_executor import CustomEventletPoolExecutor, check_evenlet_monkey_patch, evenlet_timeout_deco
from app.utils_ydf.custom_gevent_pool_executor import GeventPoolExecutor, check_gevent_monkey_patch, gevent_timeout_deco
from app import config as app_config # LogManager('pika').get_logger_and_add_handlers(10)
# LogManager('pika.heartbeat').get_logger_and_add_handlers(10)
# LogManager('rabbitpy').get_logger_and_add_handlers(10)
# LogManager('rabbitpy.base').get_logger_and_add_handlers(10)
from app.utils_ydf.custom_threadpool_executor import check_not_monkey def delete_keys_from_dict(dictx: dict, keys: list):
for dict_key in keys:
dictx.pop(dict_key) def delete_keys_and_return_new_dict(dictx: dict, keys: list):
dict_new = copy.copy(dictx) # 主要是去掉一级键 publish_time，浅拷贝即可。
for dict_key in keys:
try:
dict_new.pop(dict_key)
except KeyError:
pass
return dict_new class ExceptionForRetry(Exception):
"""为了重试的，抛出错误。只是定义了一个子类，用不用都可以""" class ExceptionForRequeue(Exception):
"""框架检测到此错误，重新放回队列中""" class ExceptionForRabbitmqRequeue(ExceptionForRequeue): # 以后去掉这个异常，抛出上面那个异常就可以了。
"""遇到此错误，重新放回队列中""" class RabbitmqClientRabbitPy:
"""
使用rabbitpy包。
""" # noinspection PyUnusedLocal
def __init__(self, username, password, host, port, virtual_host, heartbeat=0):
rabbit_url = f'amqp://{username}:{password}@{host}:{port}/{virtual_host}?heartbeat={heartbeat}'
self.connection = rabbitpy.Connection(rabbit_url) def creat_a_channel(self) -> rabbitpy.AMQP:
return rabbitpy.AMQP(self.connection.channel()) # 使用适配器，使rabbitpy包的公有方法几乎接近pika包的channel的方法。 class RabbitmqClientPika:
"""
使用pika包,多线程不安全的包。
""" def __init__(self, username, password, host, port, virtual_host, heartbeat=0):
"""
parameters = pika.URLParameters('amqp://guest:guest@localhost:5672/%2F') connection = pika.SelectConnection(parameters=parameters,
on_open_callback=on_open)
:param username:
:param password:
:param host:
:param port:
:param virtual_host:
:param heartbeat:
"""
credentials = pika.PlainCredentials(username, password)
self.connection = pika.BlockingConnection(pika.ConnectionParameters(
host, port, virtual_host, credentials, heartbeat=heartbeat))
# self.connection = pika.SelectConnection(pika.ConnectionParameters(
# host, port, virtual_host, credentials, heartbeat=heartbeat)) def creat_a_channel(self) -> BlockingChannel:
return self.connection.channel() class RabbitMqFactory:
def __init__(self, username=app_config.RABBITMQ_USER, password=app_config.RABBITMQ_PASS, host=app_config.RABBITMQ_HOST, port=app_config.RABBITMQ_PORT, virtual_host=app_config.RABBITMQ_VIRTUAL_HOST, heartbeat=60 * 10, is_use_rabbitpy=0):
"""
:param username:
:param password:
:param port:
:param virtual_host:
:param heartbeat:
:param is_use_rabbitpy: 为0使用pika，多线程不安全。为1使用rabbitpy，多线程安全的包。
"""
if is_use_rabbitpy:
self.rabbit_client = RabbitmqClientRabbitPy(username, password, host, port, virtual_host, heartbeat)
else:
self.rabbit_client = RabbitmqClientPika(username, password, host, port, virtual_host, heartbeat) def get_rabbit_cleint(self):
return self.rabbit_client class AbstractPublisher(LoggerLevelSetterMixin, metaclass=abc.ABCMeta, ):
has_init_broker = 0 def __init__(self, queue_name, log_level_int=10, logger_prefix='', is_add_file_handler=True, clear_queue_within_init=False, is_add_publish_time=False, ):
"""
:param queue_name:
:param log_level_int:
:param logger_prefix:
:param is_add_file_handler:
:param clear_queue_within_init:
"""
self._queue_name = queue_name
if logger_prefix != '':
logger_prefix += '--'
logger_name = f'{logger_prefix}{self.__class__.__name__}--{queue_name}'
self.logger = LogManager(logger_name).get_logger_and_add_handlers(log_level_int, log_filename=f'{logger_name}.log' if is_add_file_handler else None) #
# self.rabbit_client = RabbitMqFactory(is_use_rabbitpy=is_use_rabbitpy).get_rabbit_cleint()
# self.channel = self.rabbit_client.creat_a_channel()
# self.queue = self.channel.queue_declare(queue=queue_name, durable=True)
self._lock_for_pika = Lock()
self._lock_for_count = Lock()
self._current_time = None
self.count_per_minute = None
self._init_count()
self.custom_init()
self.logger.info(f'{self.__class__} 被实例化了')
self.publish_msg_num_total = 0
self._is_add_publish_time = is_add_publish_time
# atexit.register(self.__at_exit)
if clear_queue_within_init:
self.clear() def set_is_add_publish_time(self, is_add_publish_time=True):
self._is_add_publish_time = is_add_publish_time
return self def _init_count(self):
with self._lock_for_count:
self._current_time = time.time()
self.count_per_minute = 0 def custom_init(self):
pass def publish(self, msg: typing.Union[str, dict]):
if isinstance(msg, str):
msg = json.loads(msg)
if self._is_add_publish_time:
# msg.update({'publish_time': time.time(), 'publish_time_format': time_util.DatetimeConverter().datetime_str})
msg.update({'publish_time': round(time.time(), 4), })
t_start = time.time()
decorators.handle_exception(retry_times=10, is_throw_error=True, time_sleep=0.1)(self.concrete_realization_of_publish)(json.dumps(msg))
self.logger.debug(f'向{self._queue_name} 队列，推送消息 耗时{round(time.time() - t_start, 4)}秒 {msg}')
with self._lock_for_count:
self.count_per_minute += 1
self.publish_msg_num_total += 1
if time.time() - self._current_time > 10:
self.logger.info(f'10秒内推送了 {self.count_per_minute} 条消息,累计推送了 {self.publish_msg_num_total} 条消息到 {self._queue_name} 中')
self._init_count() @abc.abstractmethod
def concrete_realization_of_publish(self, msg):
raise NotImplementedError @abc.abstractmethod
def clear(self):
raise NotImplementedError @abc.abstractmethod
def get_message_count(self):
raise NotImplementedError @abc.abstractmethod
def close(self):
raise NotImplementedError def __enter__(self):
return self def __exit__(self, exc_type, exc_val, exc_tb):
self.close()
self.logger.warning(f'with中自动关闭publisher连接，累计推送了 {self.publish_msg_num_total} 条消息 ') def __at_exit(self):
self.logger.warning(f'程序关闭前，累计推送了 {self.publish_msg_num_total} 条消息 到 {self._queue_name} 中') def deco_mq_conn_error(f):
@wraps(f)
def _deco_mq_conn_error(self, *args, **kwargs):
if not self.has_init_broker:
self.logger.warning(f'对象的方法 【{f.__name__}】 首次使用 rabbitmq channel,进行初始化执行 init_broker 方法')
self.init_broker()
self.has_init_broker = 1
return f(self, *args, **kwargs)
# noinspection PyBroadException
try:
return f(self, *args, **kwargs)
except (pika.exceptions.AMQPError, amqpstorm.AMQPError) as e: # except Exception as e: # 现在装饰器用到了绝大多出地方，单个异常类型不行。ex
self.logger.error(f'rabbitmq链接出错 ,方法 {f.__name__} 出错 ，{e}')
self.init_broker()
return f(self, *args, **kwargs) return _deco_mq_conn_error class RabbitmqPublisher(AbstractPublisher):
"""
使用pika实现的。
""" # noinspection PyAttributeOutsideInit
def init_broker(self):
self.logger.warning(f'使用pika 链接mq')
self.rabbit_client = RabbitMqFactory(is_use_rabbitpy=0).get_rabbit_cleint()
self.channel = self.rabbit_client.creat_a_channel()
self.queue = self.channel.queue_declare(queue=self._queue_name, durable=True) # noinspection PyAttributeOutsideInit
@deco_mq_conn_error
def concrete_realization_of_publish(self, msg):
with self._lock_for_pika: # 亲测pika多线程publish会出错
self.channel.basic_publish(exchange='',
routing_key=self._queue_name,
body=msg,
properties=BasicProperties(
delivery_mode=2, # make message persistent 2(1是非持久化)
)
) @deco_mq_conn_error
def clear(self):
self.channel.queue_purge(self._queue_name)
self.logger.warning(f'清除 {self._queue_name} 队列中的消息成功') @deco_mq_conn_error
def get_message_count(self):
with self._lock_for_pika:
queue = self.channel.queue_declare(queue=self._queue_name, durable=True)
return queue.method.message_count # @deco_mq_conn_error
def close(self):
self.channel.close()
self.rabbit_client.connection.close()
self.logger.warning('关闭pika包 链接') class RabbitmqPublisherUsingRabbitpy(AbstractPublisher):
"""
使用rabbitpy包实现的。
""" # noinspection PyAttributeOutsideInit
def init_broker(self):
self.logger.warning(f'使用rabbitpy包 链接mq')
self.rabbit_client = RabbitMqFactory(is_use_rabbitpy=1).get_rabbit_cleint()
self.channel = self.rabbit_client.creat_a_channel()
self.queue = self.channel.queue_declare(queue=self._queue_name, durable=True) # @decorators.tomorrow_threads(10)
@deco_mq_conn_error
def concrete_realization_of_publish(self, msg):
# noinspection PyTypeChecker
self.channel.basic_publish(
exchange='',
routing_key=self._queue_name,
body=msg,
properties={'delivery_mode': 2},
) @deco_mq_conn_error
def clear(self):
self.channel.queue_purge(self._queue_name)
self.logger.warning(f'清除 {self._queue_name} 队列中的消息成功') @deco_mq_conn_error
def get_message_count(self):
# noinspection PyUnresolvedReferences
ch_raw_rabbity = self.channel.channel
return rabbitpy.amqp_queue.Queue(ch_raw_rabbity, self._queue_name, durable=True) # @deco_mq_conn_error
def close(self):
self.channel.close()
self.rabbit_client.connection.close()
self.logger.warning('关闭rabbitpy包 链接mq') class RabbitmqPublisherUsingAmqpStorm(AbstractPublisher):
# 使用amqpstorm包实现的mq操作。
# 实例属性没在init里面写，造成补全很麻烦，写在这里做类属性，方便pycharm补全
connection = amqpstorm.UriConnection
channel = amqpstorm.Channel
channel_wrapper_by_ampqstormbaic = AmqpStormBasic
queue = AmqpStormQueue # noinspection PyAttributeOutsideInit
# @decorators.synchronized
def init_broker(self):
# username=app_config.RABBITMQ_USER, password=app_config.RABBITMQ_PASS, host=app_config.RABBITMQ_HOST, port=app_config.RABBITMQ_PORT, virtual_host=app_config.RABBITMQ_VIRTUAL_HOST, heartbeat=60 * 10
self.logger.warning(f'使用AmqpStorm包 链接mq')
self.connection = amqpstorm.UriConnection(
f'amqp://{app_config.RABBITMQ_USER}:{app_config.RABBITMQ_PASS}@{app_config.RABBITMQ_HOST}:{app_config.RABBITMQ_PORT}/{app_config.RABBITMQ_VIRTUAL_HOST}?heartbeat={60 * 10}'
)
self.channel = self.connection.channel() # type:amqpstorm.Channel
self.channel_wrapper_by_ampqstormbaic = AmqpStormBasic(self.channel)
self.queue = AmqpStormQueue(self.channel)
self.queue.declare(queue=self._queue_name, durable=True) # @decorators.tomorrow_threads(10)
@deco_mq_conn_error
def concrete_realization_of_publish(self, msg):
self.channel_wrapper_by_ampqstormbaic.publish(exchange='',
routing_key=self._queue_name,
body=msg,
properties={'delivery_mode': 2}, )
# nb_print(msg) @deco_mq_conn_error
def clear(self):
self.queue.purge(self._queue_name)
self.logger.warning(f'清除 {self._queue_name} 队列中的消息成功') @deco_mq_conn_error
def get_message_count(self):
# noinspection PyUnresolvedReferences
return self.queue.declare(queue=self._queue_name, durable=True)['message_count'] # @deco_mq_conn_error
def close(self):
self.channel.close()
self.connection.close()
self.logger.warning('关闭rabbitpy包 链接mq') class RedisPublisher(AbstractPublisher, RedisMixin):
"""
使用redis作为中间件
""" def concrete_realization_of_publish(self, msg):
# noinspection PyTypeChecker
self.redis_db7.rpush(self._queue_name, msg) def clear(self):
self.redis_db7.delete(self._queue_name)
self.logger.warning(f'清除 {self._queue_name} 队列中的消息成功') def get_message_count(self):
# nb_print(self.redis_db7,self._queue_name)
return self.redis_db7.llen(self._queue_name) def close(self):
# self.redis_db7.connection_pool.disconnect()
pass class MongoMqPublisher(AbstractPublisher, MongoMixin):
# 使用mongo-queue包实现的基于mongodb的队列。
# noinspection PyAttributeOutsideInit
def custom_init(self):
self.queue = MongoQueue(
self.mongo_16_client.get_database('conqume_queues').get_collection(self._queue_name),
consumer_id=f"consumer-{time_util.DatetimeConverter().datetime_str}",
timeout=600,
max_attempts=3,
ttl=0) def concrete_realization_of_publish(self, msg):
# noinspection PyTypeChecker
self.queue.put(json.loads(msg)) def clear(self):
self.queue.clear()
self.logger.warning(f'清除 mongo队列 {self._queue_name} 中的消息成功') def get_message_count(self):
return self.queue.size() def close(self):
pass class PersistQueuePublisher(AbstractPublisher):
"""
使用persistqueue实现的本地持久化队列。
这个是本地持久化，支持本地多个启动的python脚本共享队列任务。与LocalPythonQueuePublisher相比，不会随着python解释器退出，导致任务丢失。
""" # noinspection PyAttributeOutsideInit
def custom_init(self):
# noinspection PyShadowingNames
def _my_new_db_connection(self, path, multithreading, timeout): # 主要是改了sqlite文件后缀，方便pycharm识别和打开。
# noinspection PyUnusedLocal
conn = None
if path == self._MEMORY:
conn = sqlite3.connect(path,
check_same_thread=not multithreading)
else:
conn = sqlite3.connect('{}/data.sqlite'.format(path),
timeout=timeout,
check_same_thread=not multithreading)
conn.execute('PRAGMA journal_mode=WAL;')
return conn persistqueue.SQLiteAckQueue._new_db_connection = _my_new_db_connection # 打猴子补丁。
# REMIND 官方测试基于sqlite的本地持久化，比基于纯文件的持久化，使用相同固态硬盘和操作系统情况下，速度快3倍以上，所以这里选用sqlite方式。 self.queue = persistqueue.SQLiteAckQueue(path='/sqllite_queues', name=self._queue_name, auto_commit=True, serializer=json, multithreading=True) def concrete_realization_of_publish(self, msg):
# noinspection PyTypeChecker
self.queue.put(msg) # noinspection PyProtectedMember
def clear(self):
sql = f'{"DELETE"} {"FROM"} ack_queue_{self._queue_name}'
self.logger.info(sql)
self.queue._getter.execute(sql)
self.queue._getter.commit()
self.logger.warning(f'清除 本地持久化队列 {self._queue_name} 中的消息成功') def get_message_count(self):
return self.queue.qsize() def close(self):
pass local_pyhton_queue_name__local_pyhton_queue_obj_map = dict() # 使local queue和其他中间件完全一样的使用方式，使用映射保存队列的名字，使消费和发布通过队列名字能找到队列对象。 class LocalPythonQueuePublisher(AbstractPublisher):
"""
使用redis作为中间件
""" # noinspection PyAttributeOutsideInit
def custom_init(self):
if self._queue_name not in local_pyhton_queue_name__local_pyhton_queue_obj_map:
local_pyhton_queue_name__local_pyhton_queue_obj_map[self._queue_name] = Queue()
self.queue = local_pyhton_queue_name__local_pyhton_queue_obj_map[self._queue_name] def concrete_realization_of_publish(self, msg):
# noinspection PyTypeChecker
self.queue.put(msg) def clear(self):
# noinspection PyUnresolvedReferences
self.queue.queue.clear()
self.logger.warning(f'清除 本地队列中的消息成功') def get_message_count(self):
return self.queue.qsize() def close(self):
pass class RedisFilter(RedisMixin):
def __init__(self, redis_key_name):
self._redis_key_name = redis_key_name @staticmethod
def _get_ordered_str(value):
"""对json的键值对在redis中进行过滤，需要先把键值对排序，否则过滤会不准确如 {"a":1,"b":2} 和 {"b":2,"a":1}"""
if isinstance(value, str):
value = json.loads(value)
ordered_dict = OrderedDict()
for k in sorted(value):
ordered_dict[k] = value[k]
return json.dumps(ordered_dict) def add_a_value(self, value: typing.Union[str, dict]):
self.redis_db7.sadd(self._redis_key_name, self._get_ordered_str(value)) def check_value_exists(self, value):
return self.redis_db7.sismember(self._redis_key_name, self._get_ordered_str(value)) class AbstractConsumer(LoggerLevelSetterMixin, metaclass=abc.ABCMeta, ):
time_interval_for_check_do_not_run_time = 60
BROKER_KIND = None @property
@decorators.synchronized
def publisher_of_same_queue(self):
if not self._publisher_of_same_queue:
self._publisher_of_same_queue = get_publisher(self._queue_name, broker_kind=self.BROKER_KIND)
if self._msg_expire_senconds:
self._publisher_of_same_queue.set_is_add_publish_time()
return self._publisher_of_same_queue @classmethod
def join_shedual_task_thread(cls):
""" :return:
"""
"""
def ff():
RabbitmqConsumer('queue_test', consuming_function=f3, threads_num=20, msg_schedule_time_intercal=2, log_level=10, logger_prefix='yy平台消费', is_consuming_function_use_multi_params=True).start_consuming_message()
RabbitmqConsumer('queue_test2', consuming_function=f4, threads_num=20, msg_schedule_time_intercal=4, log_level=10, logger_prefix='zz平台消费', is_consuming_function_use_multi_params=True).start_consuming_message()
AbstractConsumer.join_shedual_task_thread() # 如果开多进程启动消费者，在linux上需要这样写下这一行。 if __name__ == '__main__':
[Process(target=ff).start() for _ in range(4)] """
ConcurrentModeDispatcher.join() def __init__(self, queue_name, *, consuming_function: Callable = None, function_timeout=0, threads_num=50, specify_threadpool=None, concurrent_mode=1,
max_retry_times=3, log_level=10, is_print_detail_exception=True, msg_schedule_time_intercal=0.0, msg_expire_senconds=0,
logger_prefix='', create_logger_file=True, do_task_filtering=False, is_consuming_function_use_multi_params=True,
is_do_not_run_by_specify_time_effect=False, do_not_run_by_specify_time=('10:00:00', '22:00:00'), schedule_tasks_on_main_thread=False):
"""
:param queue_name:
:param consuming_function: 处理消息的函数。
:param function_timeout : 超时秒数，函数运行超过这个时间，则自动杀死函数。为0是不限制。
:param threads_num:
:param specify_threadpool:使用指定的线程池，可以多个消费者共使用一个线程池，不为None时候。threads_num失效
:param concurrent_mode:并发模式，暂时支持 线程 、gevent、eventlet三种模式。 1线程 2 gevent 3 evenlet
:param max_retry_times:
:param log_level:
:param is_print_detail_exception:
:param msg_schedule_time_intercal:消息调度的时间间隔，用于控频
:param logger_prefix: 日志前缀，可使不同的消费者生成不同的日志
:param create_logger_file : 是否创建文件日志
:param do_task_filtering :是否执行基于函数参数的任务过滤
:is_consuming_function_use_multi_params 函数的参数是否是传统的多参数，不为单个body字典表示多个参数。
:param is_do_not_run_by_specify_time_effect :是否使不运行的时间段生效
:param do_not_run_by_specify_time :不运行的时间段
:param schedule_tasks_on_main_thread :直接在主线程调度任务，意味着不能直接在当前主线程同时开启两个消费者。
"""
self._queue_name = queue_name
self.queue_name = queue_name # 可以换成公有的，免得外部访问有警告。
self.consuming_function = consuming_function
self._function_timeout = function_timeout
self._threads_num = threads_num
self._specify_threadpool = specify_threadpool
self._threadpool = None # 单独加一个检测消息数量和心跳的线程
self._concurrent_mode = concurrent_mode
self._max_retry_times = max_retry_times
self._is_print_detail_exception = is_print_detail_exception
self._msg_schedule_time_intercal = msg_schedule_time_intercal if msg_schedule_time_intercal > 0.001 else 0.001
self._msg_expire_senconds = msg_expire_senconds if self._concurrent_mode not in (1, 2, 3):
raise ValueError('设置的并发模式不正确')
self._concurrent_mode_dispatcher = ConcurrentModeDispatcher(self) self._logger_prefix = logger_prefix
self._log_level = log_level
if logger_prefix != '':
logger_prefix += '--'
logger_name = f'{logger_prefix}{self.__class__.__name__}--{self._concurrent_mode_dispatcher.concurrent_name}--{queue_name}'
# nb_print(logger_name)
self.logger = LogManager(logger_name).get_logger_and_add_handlers(log_level, log_filename=f'{logger_name}.log' if create_logger_file else None)
self.logger.info(f'{self.__class__} 被实例化') self._do_task_filtering = do_task_filtering
self._redis_filter_key_name = f'filter:{queue_name}'
self._redis_filter = RedisFilter(self._redis_filter_key_name) self._is_consuming_function_use_multi_params = is_consuming_function_use_multi_params
self._lock_for_pika = Lock() self._execute_task_times_every_minute = 0 # 每分钟执行了多少次任务。
self._lock_for_count_execute_task_times_every_minute = Lock()
self._current_time_for_execute_task_times_every_minute = time.time() self._msg_num_in_broker = 0
self._last_timestamp_when_has_task_in_queue = 0
self._last_timestamp_print_msg_num = 0 self._is_do_not_run_by_specify_time_effect = is_do_not_run_by_specify_time_effect
self._do_not_run_by_specify_time = do_not_run_by_specify_time # 可以设置在指定的时间段不运行。
self._schedule_tasks_on_main_thread = schedule_tasks_on_main_thread self.stop_flag = False self._publisher_of_same_queue = None @property
@decorators.synchronized
def threadpool(self):
return self._concurrent_mode_dispatcher.build_pool() def keep_circulating(self, time_sleep=0.001, exit_if_function_run_sucsess=False, is_display_detail_exception=True):
"""间隔一段时间，一直循环运行某个方法的装饰器
:param time_sleep :循环的间隔时间
:param is_display_detail_exception
:param exit_if_function_run_sucsess :如果成功了就退出循环
""" def _keep_circulating(func):
# noinspection PyBroadException
@wraps(func)
def __keep_circulating(*args, **kwargs):
while 1:
if self.stop_flag:
break
try:
result = func(*args, **kwargs)
if exit_if_function_run_sucsess:
return result
except Exception as e:
msg = func.__name__ + ' 运行出错\n ' + traceback.format_exc(limit=10) if is_display_detail_exception else str(e)
self.logger.error(msg)
finally:
time.sleep(time_sleep) return __keep_circulating return _keep_circulating def start_consuming_message(self):
self.logger.warning(f'开始消费 {self._queue_name} 中的消息')
# self.threadpool.submit(decorators.keep_circulating(20)(self.check_heartbeat_and_message_count))
self.threadpool.submit(self.keep_circulating(20)(self.check_heartbeat_and_message_count))
if self._schedule_tasks_on_main_thread:
# decorators.keep_circulating(1)(self._shedual_task)()
self.keep_circulating(1)(self._shedual_task)()
else:
# t = Thread(target=decorators.keep_circulating(1)(self._shedual_task))
self._concurrent_mode_dispatcher.schedulal_task_with_no_block() @abc.abstractmethod
def _shedual_task(self):
raise NotImplementedError def _run_consuming_function_with_confirm_and_retry(self, kw: dict, current_retry_times=0):
if self._do_task_filtering and self._redis_filter.check_value_exists(kw['body']): # 对函数的参数进行检查，过滤已经执行过并且成功的任务。
self.logger.info(f'redis的 [{self._redis_filter_key_name}] 键 中 过滤任务 {kw["body"]}')
self._confirm_consume(kw)
return
with self._lock_for_count_execute_task_times_every_minute:
self._execute_task_times_every_minute += 1
if time.time() - self._current_time_for_execute_task_times_every_minute > 60:
self.logger.info(
f'一分钟内执行了 {self._execute_task_times_every_minute} 次函数 [ {self.consuming_function.__name__} ] ,预计'
f'还需要 {time_util.seconds_to_hour_minute_second(self._msg_num_in_broker / self._execute_task_times_every_minute * 60)} 时间'
f'才能执行完成 {self._msg_num_in_broker}个剩余的任务 ')
self._current_time_for_execute_task_times_every_minute = time.time()
self._execute_task_times_every_minute = 0 if current_retry_times < self._max_retry_times + 1:
# noinspection PyBroadException
t_start = time.time()
try:
function_run = self.consuming_function if self._function_timeout == 0 else self._concurrent_mode_dispatcher.timeout_deco(self._function_timeout)(self.consuming_function)
if self._is_consuming_function_use_multi_params: # 消费函数使用传统的多参数形式
function_run(**delete_keys_and_return_new_dict(kw['body'], ['publish_time', 'publish_time_format']))
else:
function_run(delete_keys_and_return_new_dict(kw['body'], ['publish_time', 'publish_time_format'])) # 消费函数使用单个参数，参数自身是一个字典，由键值对表示各个参数。
self._confirm_consume(kw)
if self._do_task_filtering:
self._redis_filter.add_a_value(kw['body']) # 函数执行成功后，添加函数的参数排序后的键值对字符串到set中。 self.logger.debug(f'{self._concurrent_mode_dispatcher.get_concurrent_info()} 函数 {self.consuming_function.__name__} '
f'第{current_retry_times + 1}次 运行, 正确了，函数运行时间是 {round(time.time() - t_start, 4)} 秒,入参是 【 {kw["body"]} 】')
except Exception as e:
if isinstance(e, (PyMongoError, ExceptionForRequeue)): # mongo经常维护备份时候插入不了或挂了，或者自己主动抛出一个ExceptionForRequeue类型的错误会重新入队，不受指定重试次数逇约束。
self.logger.critical(f'函数 [{self.consuming_function.__name__}] 中发生错误 {type(e)} {e}')
return self._requeue(kw)
self.logger.error(f'函数 {self.consuming_function.__name__} 第{current_retry_times + 1}次发生错误，'
f'函数运行时间是 {round(time.time() - t_start, 4)} 秒,\n 入参是 【 {kw["body"]} 】 \n 原因是 {type(e)} {e} ', exc_info=self._is_print_detail_exception)
self._run_consuming_function_with_confirm_and_retry(kw, current_retry_times + 1)
else:
self.logger.critical(f'函数 {self.consuming_function.__name__} 达到最大重试次数 {self._max_retry_times} 后,仍然失败， 入参是 【 {kw["body"]} 】') # 错得超过指定的次数了，就确认消费了。
self._confirm_consume(kw) @abc.abstractmethod
def _confirm_consume(self, kw):
"""确认消费"""
raise NotImplementedError # noinspection PyUnusedLocal def check_heartbeat_and_message_count(self):
self._msg_num_in_broker = self.publisher_of_same_queue.get_message_count()
if time.time() - self._last_timestamp_print_msg_num > 60:
self.logger.info(f'[{self._queue_name}] 队列中还有 [{self._msg_num_in_broker}] 个任务')
self._last_timestamp_print_msg_num = time.time()
if self._msg_num_in_broker != 0:
self._last_timestamp_when_has_task_in_queue = time.time()
return self._msg_num_in_broker @abc.abstractmethod
def _requeue(self, kw):
"""重新入队"""
raise NotImplementedError def _submit_task(self, kw):
if self._judge_is_daylight():
self._requeue(kw)
time.sleep(self.time_interval_for_check_do_not_run_time)
return
if self._msg_expire_senconds != 0 and time.time() - self._msg_expire_senconds > kw['body']['publish_time']:
self.logger.warning(f'消息发布时戳是 {kw["body"]["publish_time"]} {kw["body"].get("publish_time_format", "")},距离现在 {round(time.time() - kw["body"]["publish_time"], 4)} 秒 ,'
f'超过了指定的 {self._msg_expire_senconds} 秒，丢弃任务')
self._confirm_consume(kw)
return 0
self.threadpool.submit(self._run_consuming_function_with_confirm_and_retry, kw)
time.sleep(self._msg_schedule_time_intercal) def _judge_is_daylight(self):
if self._is_do_not_run_by_specify_time_effect and self._do_not_run_by_specify_time[0] < time_util.DatetimeConverter().time_str < self._do_not_run_by_specify_time[1]:
self.logger.warning(f'现在时间是 {time_util.DatetimeConverter()} ，现在时间是在 {self._do_not_run_by_specify_time} 之间，不运行')
return True def __str__(self):
return f'队列为 {self.queue_name} 函数为 {self.consuming_function} 的消费者' # noinspection PyProtectedMember
class ConcurrentModeDispatcher(LoggerMixin):
schedulal_thread_to_be_join = []
concurrent_mode = None
schedual_task_always_use_thread = False def __init__(self, consumerx: AbstractConsumer):
self.consumer = consumerx
if self.__class__.concurrent_mode is not None and self.consumer._concurrent_mode != self.__class__.concurrent_mode:
raise ValueError('同一解释器中不可以设置两种并发类型')
self._concurrent_mode = self.__class__.concurrent_mode = self.consumer._concurrent_mode
concurrent_name = ''
self.timeout_deco = None
if self._concurrent_mode == 1:
concurrent_name = 'thread'
self.timeout_deco = decorators.timeout
elif self._concurrent_mode == 2:
concurrent_name = 'gevent'
self.timeout_deco = gevent_timeout_deco
elif self._concurrent_mode == 3:
concurrent_name = 'evenlet'
self.timeout_deco = evenlet_timeout_deco
self.concurrent_name = concurrent_name
self.logger.warning(f'{self.consumer} 设置并发模式为 {self.concurrent_name}') def build_pool(self):
if self.consumer._threadpool:
return self.consumer._threadpool pool_type = None # 是按照ThreadpoolExecutor写的三个鸭子类，公有方法名和功能写成完全一致，可以互相替换。
if self._concurrent_mode == 1:
pool_type = CustomThreadPoolExecutor
check_not_monkey()
elif self._concurrent_mode == 2:
pool_type = GeventPoolExecutor
check_gevent_monkey_patch()
elif self._concurrent_mode == 3:
pool_type = CustomEventletPoolExecutor
check_evenlet_monkey_patch()
self.consumer._threadpool = self.consumer._specify_threadpool if self.consumer._specify_threadpool else pool_type(self.consumer._threads_num + 1) # 单独加一个检测消息数量和心跳的线程
self.logger.warning(f'{self.concurrent_name} {self.consumer._threadpool}')
return self.consumer._threadpool def schedulal_task_with_no_block(self):
if self.schedual_task_always_use_thread:
t = Thread(target=self.consumer.keep_circulating(1)(self.consumer._shedual_task))
self.__class__.schedulal_thread_to_be_join.append(t)
t.start()
else:
if self._concurrent_mode == 1:
t = Thread(target=self.consumer.keep_circulating(1)(self.consumer._shedual_task))
self.__class__.schedulal_thread_to_be_join.append(t)
t.start()
elif self._concurrent_mode == 2:
g = gevent.spawn(self.consumer.keep_circulating(1)(self.consumer._shedual_task), )
self.__class__.schedulal_thread_to_be_join.append(g)
elif self._concurrent_mode == 3:
g = eventlet.spawn(self.consumer.keep_circulating(1)(self.consumer._shedual_task), )
self.__class__.schedulal_thread_to_be_join.append(g)
atexit.register(self.join) @classmethod
def join(cls):
nb_print((cls.schedulal_thread_to_be_join, len(cls.schedulal_thread_to_be_join), '模式：', cls.concurrent_mode))
if cls.schedual_task_always_use_thread:
for t in cls.schedulal_thread_to_be_join:
nb_print(t)
t.join()
else:
if cls.concurrent_mode == 1:
for t in cls.schedulal_thread_to_be_join:
nb_print(t)
t.join()
elif cls.concurrent_mode == 2:
# cls.logger.info()
nb_print(cls.schedulal_thread_to_be_join)
gevent.joinall(cls.schedulal_thread_to_be_join, raise_error=True, )
elif cls.concurrent_mode == 3:
for g in cls.schedulal_thread_to_be_join:
# eventlet.greenthread.GreenThread.
nb_print(g)
g.wait() def get_concurrent_info(self):
concurrent_info = ''
if self._concurrent_mode == 1:
concurrent_info = f'[{threading.current_thread()} {threading.active_count()}]'
elif self._concurrent_mode == 2:
concurrent_info = f'[{gevent.getcurrent()} {threading.active_count()}]'
elif self._concurrent_mode == 3:
# noinspection PyArgumentList
concurrent_info = f'[{eventlet.getcurrent()} {threading.active_count()}]'
return concurrent_info def wait_for_possible_has_finish_all_tasks(queue_name: str, minutes: int, send_stop_to_broker=0, broker_kind: int = 0, ):
"""
由于是异步消费，和存在队列一边被消费，一边在推送，或者还有结尾少量任务还在确认消费者实际还没彻底运行完成。 但有时候需要判断 所有任务，务是否完成，提供一个不精确的判断，要搞清楚原因和场景后再慎用。
:param queue_name: 队列名字
:param minutes: 连续多少分钟没任务就判断为消费已完成
:param send_stop_to_broker :发送停止标志到中间件，这回导致消费退出循环调度。
:param broker_kind: 中间件种类
:return:
"""
if minutes <= 1:
raise ValueError('疑似完成任务，判断时间最少需要设置为2分钟内,最好是是10分钟')
pb = get_publisher(queue_name, broker_kind=broker_kind)
no_task_time = 0
while 1:
# noinspection PyBroadException
try:
message_count = pb.get_message_count()
except Exception as e:
nb_print(e)
message_count = -1
if message_count == 0:
no_task_time += 30
else:
no_task_time = 0
time.sleep(30)
if no_task_time > minutes * 60:
break
if send_stop_to_broker:
pb.publish({'stop': 1})
pb.close() class RabbitmqConsumer(AbstractConsumer):
"""
使用pika包实现的。
"""
BROKER_KIND = 0 def _shedual_task(self):
channel = RabbitMqFactory(is_use_rabbitpy=0).get_rabbit_cleint().creat_a_channel()
channel.queue_declare(queue=self._queue_name, durable=True)
channel.basic_qos(prefetch_count=self._threads_num) def callback(ch, method, properties, body):
body = body.decode()
self.logger.debug(f'从rabbitmq的 [{self._queue_name}] 队列中 取出的消息是： {body}')
body = json.loads(body)
kw = {'ch': ch, 'method': method, 'properties': properties, 'body': body}
self._submit_task(kw) channel.basic_consume(callback,
queue=self._queue_name,
# no_ack=True
)
channel.start_consuming() def _confirm_consume(self, kw):
with self._lock_for_pika:
try:
kw['ch'].basic_ack(delivery_tag=kw['method'].delivery_tag) # 确认消费
except pika.exceptions.AMQPError as e:
self.logger.error(f'pika确认消费失败 {e}') def _requeue(self, kw):
with self._lock_for_pika:
# ch.connection.add_callback_threadsafe(functools.partial(self.__ack_message_pika, ch, method.delivery_tag))
return kw['ch'].basic_nack(delivery_tag=kw['method'].delivery_tag) # 立即重新入队。 @staticmethod
def __ack_message_pika(channelx, delivery_tagx):
"""Note that `channel` must be the same pika channel instance via which
the message being ACKed was retrieved (AMQP protocol constraint).
"""
if channelx.is_open:
channelx.basic_ack(delivery_tagx)
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.
pass class RabbitmqConsumerAmqpStorm(AbstractConsumer):
"""
使用AmqpStorm实现的，多线程安全的，不用加锁。
"""
BROKER_KIND = 4 def _shedual_task(self):
# noinspection PyTypeChecker
def callback(amqpstorm_message: amqpstorm.Message):
body = amqpstorm_message.body
self.logger.debug(f'从rabbitmq的 [{self._queue_name}] 队列中 取出的消息是： {body}')
body = json.loads(body)
kw = {'amqpstorm_message': amqpstorm_message, 'body': body}
self._submit_task(kw) rp = RabbitmqPublisherUsingAmqpStorm(self.queue_name)
rp.init_broker()
rp.channel_wrapper_by_ampqstormbaic.qos(self._threads_num)
rp.channel_wrapper_by_ampqstormbaic.consume(callback=callback, queue=self.queue_name, no_ack=False)
rp.channel.start_consuming(auto_decode=True) def _confirm_consume(self, kw):
# noinspection PyBroadException
try:
kw['amqpstorm_message'].ack() # 确认消费
except Exception as e:
self.logger.error(f'AmqpStorm确认消费失败 {type(e)} {e}') def _requeue(self, kw):
kw['amqpstorm_message'].nack(requeue=True) class RabbitmqConsumerRabbitpy(AbstractConsumer):
"""
使用rabbitpy实现的
"""
BROKER_KIND = 1 def _shedual_task(self):
# noinspection PyTypeChecker
channel = RabbitMqFactory(is_use_rabbitpy=1).get_rabbit_cleint().creat_a_channel() # type: rabbitpy.AMQP #
channel.queue_declare(queue=self._queue_name, durable=True)
channel.basic_qos(prefetch_count=self._threads_num)
for message in channel.basic_consume(self._queue_name, no_ack=False):
body = message.body.decode()
self.logger.debug(f'从rabbitmq {self._queue_name} 队列中 取出的消息是： {body}')
kw = {'message': message, 'body': json.loads(message.body.decode())}
self._submit_task(kw) def _confirm_consume(self, kw):
kw['message'].ack() def _requeue(self, kw):
kw['message'].nack(requeue=True) class RedisConsumer(AbstractConsumer, RedisMixin):
"""
redis作为中间件实现的。
"""
BROKER_KIND = 2 def _shedual_task(self):
while True:
t_start = time.time()
task_bytes = self.redis_db7.blpop(self._queue_name)[1] # 使用db7
if task_bytes:
self.logger.debug(f'取出的任务时间是 {round(time.time() - t_start, 4)} 消息是： {task_bytes.decode()} ')
task_dict = json.loads(task_bytes)
kw = {'body': task_dict}
self._submit_task(kw) def _confirm_consume(self, kw):
pass # redis没有确认消费的功能。 def _requeue(self, kw):
self.redis_db7.rpush(self._queue_name, json.dumps(kw['body'])) class MongoMqConsumer(AbstractConsumer, MongoMixin):
"""
Mongo queue包实现的基于mongo的消息队列，支持消费确认。
"""
BROKER_KIND = 5 def _shedual_task(self):
mp = MongoMqPublisher(self.queue_name)
while True:
t_start = time.time()
job = mp.queue.next()
if job is not None:
self.logger.debug(f'取出的任务时间是 {round(time.time() - t_start, 4)} 消息是： {job.payload} ')
kw = {'body': job.payload, 'job': job}
self._submit_task(kw)
else:
time.sleep(self._msg_schedule_time_intercal) def _confirm_consume(self, kw):
kw['job'].complete() def _requeue(self, kw):
kw['job'].release() class PersistQueueConsumer(AbstractConsumer):
"""
persist queue包实现的本地持久化消息队列。
"""
BROKER_KIND = 6 def _shedual_task(self):
pub = PersistQueuePublisher(self.queue_name)
while True:
t_start = time.time()
item = pub.queue.get()
self.logger.debug(f'取出的任务时间是 {round(time.time() - t_start, 4)} 消息是： {item} ')
kw = {'body': json.loads(item), 'q': pub.queue, 'item': item}
self._submit_task(kw) def _confirm_consume(self, kw):
kw['q'].ack(kw['item']) def _requeue(self, kw):
kw['q'].nack(kw['item']) class LocalPythonQueueConsumer(AbstractConsumer):
BROKER_KIND = 3 @property
def local_python_queue(self) -> Queue:
return local_pyhton_queue_name__local_pyhton_queue_obj_map[self._queue_name] def _shedual_task(self):
while True:
t_start = time.time()
task = self.local_python_queue.get()
if isinstance(task, str):
task = json.loads(task)
self.logger.debug(f'取出的任务时间是 {round(time.time() - t_start, 4)} 消息是： {json.dumps(task)} ')
task_dict = task
kw = {'body': task_dict}
self._submit_task(kw) def _confirm_consume(self, kw):
pass def _requeue(self, kw):
self.local_python_queue.put(kw['body']) def get_publisher(queue_name, *, log_level_int=10, logger_prefix='', is_add_file_handler=False, clear_queue_within_init=False, is_add_publish_time=False, broker_kind=0):
"""
:param queue_name:
:param log_level_int:
:param logger_prefix:
:param is_add_file_handler:
:param clear_queue_within_init:
:param is_add_publish_time:是否添加发布时间到中间件，如果设置了过期时间不为0，需要设为True
:param broker_kind: 中间件或使用包的种类。
:return:
"""
all_kwargs = copy.deepcopy(locals())
all_kwargs.pop('broker_kind')
if broker_kind == 0:
return RabbitmqPublisher(**all_kwargs)
elif broker_kind == 1:
return RabbitmqPublisherUsingRabbitpy(**all_kwargs)
elif broker_kind == 2:
return RedisPublisher(**all_kwargs)
elif broker_kind == 3:
return LocalPythonQueuePublisher(**all_kwargs)
elif broker_kind == 4:
return RabbitmqPublisherUsingAmqpStorm(**all_kwargs)
elif broker_kind == 5:
return MongoMqPublisher(**all_kwargs)
elif broker_kind == 6:
return PersistQueuePublisher(**all_kwargs)
else:
raise ValueError('设置的中间件种类数字不正确') def get_consumer(queue_name, *, consuming_function: Callable = None, function_timeout=0, threads_num=50, specify_threadpool=None, concurrent_mode=1,
max_retry_times=3, log_level=10, is_print_detail_exception=True, msg_schedule_time_intercal=0.0, msg_expire_senconds=0,
logger_prefix='', create_logger_file=True, do_task_filtering=False, is_consuming_function_use_multi_params=True,
is_do_not_run_by_specify_time_effect=False, do_not_run_by_specify_time=('10:00:00', '22:00:00'),
schedule_tasks_on_main_thread=False, broker_kind=0):
"""
使用工厂模式再包一层，通过设置数字来生成基于不同中间件或包的consumer。
:param queue_name:
:param consuming_function: 处理消息的函数。
:param function_timeout : 超时秒数，函数运行超过这个时间，则自动杀死函数。为0是不限制。
:param threads_num:
:param specify_threadpool:使用指定的线程池，可以多个消费者共使用一个线程池，不为None时候。threads_num失效
:param concurrent_mode:并发模式，1线程 2gevent 3eventlet
:param max_retry_times:
:param log_level:
:param is_print_detail_exception:
:param msg_schedule_time_intercal:消息调度的时间间隔，用于控频
:param msg_expire_senconds:消息过期时间，为0永不过期，为10则代表，10秒之前发布的任务如果现在才轮到消费则丢弃任务。
:param logger_prefix: 日志前缀，可使不同的消费者生成不同的日志
:param create_logger_file : 是否创建文件日志
:param do_task_filtering :是否执行基于函数参数的任务过滤
:param is_consuming_function_use_multi_params 函数的参数是否是传统的多参数，不为单个body字典表示多个参数。
:param is_do_not_run_by_specify_time_effect :是否使不运行的时间段生效
:param do_not_run_by_specify_time :不运行的时间段
:param schedule_tasks_on_main_thread :直接在主线程调度任务，意味着不能直接在当前主线程同时开启两个消费者。
:param broker_kind:中间件种类,，不要设置为1。 0 使用pika链接mq，2使用redis，3使用python内置Queue
:return
"""
all_kwargs = copy.copy(locals())
all_kwargs.pop('broker_kind')
if broker_kind == 0:
return RabbitmqConsumer(**all_kwargs)
elif broker_kind == 1:
return RabbitmqConsumerRabbitpy(**all_kwargs)
elif broker_kind == 2:
return RedisConsumer(**all_kwargs)
elif broker_kind == 3:
return LocalPythonQueueConsumer(**all_kwargs)
elif broker_kind == 4:
return RabbitmqConsumerAmqpStorm(**all_kwargs)
elif broker_kind == 5:
return MongoMqConsumer(**all_kwargs)
elif broker_kind == 6:
return PersistQueueConsumer(**all_kwargs)
else:
raise ValueError('设置的中间件种类数字不正确') # noinspection PyMethodMayBeStatic,PyShadowingNames
class _Test(unittest.TestCase, LoggerMixin, RedisMixin):
"""
演示一个简单求和的例子。
""" @unittest.skip
def test_publisher_with(self):
"""
测试上下文管理器。
:return:
"""
with RabbitmqPublisher('queue_test') as rp:
for i in range(1000):
rp.publish(str(i)) @unittest.skip
def test_publish_rabbit(self):
"""
测试mq推送
:return:
"""
rabbitmq_publisher = RabbitmqPublisher('queue_test', log_level_int=10, logger_prefix='yy平台推送')
rabbitmq_publisher.clear()
for i in range(500000):
try:
time.sleep(1)
rabbitmq_publisher.publish({'a': i, 'b': 2 * i})
except Exception as e:
print(e) rabbitmq_publisher = RabbitmqPublisher('queue_test2', log_level_int=20, logger_prefix='zz平台推送')
rabbitmq_publisher.clear()
[rabbitmq_publisher.publish({'somestr_to_be_print': str(i)}) for i in range(500000)] @unittest.skip
def test_publish_redis(self):
# 如果需要批量推送
for i in range(10007):
# 最犀利的批量操作方式，自动聚合多条redis命令，支持多种redis混合命令批量操作。
RedisBulkWriteHelper(self.redis_db7, 1000).add_task(RedisOperation('lpush', 'queue_test', json.dumps({'a': i, 'b': 2 * i})))
[self.redis_db7.lpush('queue_test', json.dumps({'a': j, 'b': 2 * j})) for j in range(500)]
print('推送完毕') @unittest.skip
def test_consume(self):
"""
单参数代表所有传参
:return:
""" def f(body):
self.logger.info(f'消费此消息 {body}')
# print(body['a'] + body['b'])
time.sleep(5) # 模拟做某事需要阻塞10秒种，必须用并发。 # 把消费的函数名传给consuming_function，就这么简单。
rabbitmq_consumer = RabbitmqConsumer('queue_test', consuming_function=f, threads_num=20, msg_schedule_time_intercal=0.5, log_level=10, logger_prefix='yy平台消费',
is_consuming_function_use_multi_params=False)
rabbitmq_consumer.start_consuming_message() @unittest.skip
def test_consume2(self):
"""
测试支持传统参数形式，不是用一个字典里面包含所有参数。
:return:
""" def f2(a, b):
self.logger.debug(f'a的值是 {a}')
self.logger.debug(f'b的值是 {b}')
print(f'{a} + {b} 的和是 {a + b}')
time.sleep(3) # 模拟做某事需要阻塞10秒种，必须用并发。 # 把消费的函数名传给consuming_function，就这么简单。
RabbitmqConsumer('queue_test', consuming_function=f2, threads_num=60, msg_schedule_time_intercal=5, log_level=10, logger_prefix='yy平台消费', is_consuming_function_use_multi_params=True).start_consuming_message() @unittest.skip
def test_redis_filter(self):
"""
测试基于redis set结构的过滤器。
:return:
"""
redis_filter = RedisFilter('abcd')
redis_filter.add_a_value({'a': 1, 'c': 3, 'b': 2})
redis_filter.check_value_exists({'a': 1, 'c': 3, 'b': 2})
redis_filter.check_value_exists({'a': 1, 'b': 2, 'c': 3})
with decorators.TimerContextManager():
print(redis_filter.check_value_exists('{"a": 1, "b": 2, "c": 3}'))
with decorators.TimerContextManager():
# 实测百万元素的set，过滤检查不需要1毫秒,一般最多100万个酒店。
print(RedisFilter('filter:mafengwo-detail_task').check_value_exists({"_id": ""})) @unittest.skip
def test_run_two_function(self):
# 演示连续运行两个consumer
def f3(a, b):
print(f'{a} + {b} = {a + b}')
time.sleep(10) # 模拟做某事需要阻塞10秒种，必须用并发。 def f4(somestr_to_be_print):
print(f'打印 {somestr_to_be_print}')
time.sleep(20) # 模拟做某事需要阻塞10秒种，必须用并发。 RabbitmqConsumer('queue_test', consuming_function=f3, threads_num=20, msg_schedule_time_intercal=2, log_level=10, logger_prefix='yy平台消费', is_consuming_function_use_multi_params=True).start_consuming_message()
RabbitmqConsumer('queue_test2', consuming_function=f4, threads_num=20, msg_schedule_time_intercal=4, log_level=10, logger_prefix='zz平台消费', is_consuming_function_use_multi_params=True).start_consuming_message()
# AbstractConsumer.join_shedual_task_thread() @unittest.skip
def test_local_python_queue_as_broker(self):
def f8(x, y):
nb_print((x, y)) consumer = get_consumer('queue_testlll', consuming_function=f8, threads_num=30, msg_schedule_time_intercal=1, log_level=10, logger_prefix='uu平台消费',
function_timeout=20, is_print_detail_exception=True, msg_expire_senconds=5, broker_kind=3) # 通过设置broker_kind，一键切换中间件为mq或redis
get_publisher('queue_testlll', broker_kind=3, is_add_publish_time=True).publish({'x': 3, 'y': 4})
consumer.publisher_of_same_queue.set_is_add_publish_time(True).publish({'x': 1, 'y': 2})
nb_print(consumer.publisher_of_same_queue.get_message_count())
consumer.start_consuming_message()
for i in range(10000):
consumer.publisher_of_same_queue.publish({'x': i, 'y': i * 2})
time.sleep(2) # @unittest.skip
def test_factory_pattern_consumer(self):
"""
测试工厂模式来生成消费者
:return:
""" def f2(a, b):
# body_dict = json.loads(body) self.logger.info(f'消费此消息 {a} {b} ,结果是 {a + b}')
# print(body_dict['a'] + body_dict['b'])
time.sleep(30) # 模拟做某事需要阻塞10秒种，必须用并发。
# 把消费的函数名传给consuming_function，就这么简单。 consumer = get_consumer('queue_test5', consuming_function=f2, threads_num=30, msg_schedule_time_intercal=1, log_level=10, logger_prefix='zz平台消费',
function_timeout=20, is_print_detail_exception=True, msg_expire_senconds=500, broker_kind=0) # 通过设置broker_kind，一键切换中间件为mq或redis
consumer.publisher_of_same_queue.clear()
[consumer.publisher_of_same_queue.publish({'a': i, 'b': 2 * i}) for i in range(1)]
time.sleep(10) # sleep测试消息过期。
get_publisher('queue_test5', broker_kind=0).set_is_add_publish_time().publish({'a': 1000, 'b': 2000})
consumer.start_consuming_message()
# consumer.join_shedual_task_thread()
# block_python_exit.just_block_python_exit()
# show_current_threads_num(block=True) if __name__ == '__main__':
# noinspection PyArgumentList
unittest.main(sleep_time=1)
gevent 并发模式：
本地持久化队列，使用sqlite3模拟消息队列的图片。
mongodb模拟的消息队列
