Celery: 实现集群消费和广播消费

雨点打透心脏的1/2处 2022-12-03 13:08 233阅读 0赞

#### 集群消费 ####

集群消费类似于负载均衡，生产产生的消息只会被一个worker消费一次  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70_pic_center]

##### 代码 #####

*  worker

from kombu import Queue
    from kombu import Exchange
    from kombu.common import Broadcast
    
    import logging
    from celery import Celery
    
    # logging
    logging.basicConfig(level=logging.INFO,
                            format='%(asctime)s %(levelname)s %(message)s'
                            )
    logger = logging.getLogger(__name__)
    
    
    # mq config
    app = Celery("node_worker")
    app.conf.task_queues = [
        Queue(f'write-mysql', routing_key='write-mysql')
    ]
    app.conf.task_routes = { 
        "write-mysql": { 
            'queue': 'write-mysql',
            'routing_key': 'write-mysql'
        }
    }
    # rabbit mq的信息
    app.conf.broker_url = 'amqp://user:password@127.0.0.1:5672/celeryhost'
    
    # task
    @app.task(name=f"write-mysql")
    def write_sql_task(message):
        logger.error(message)

*  启动worker

# 启动worker 1
    celery -A workerOne worker -l info -S one/ -n one@%h
    # 启动worker 2
    celery -A workerOne worker -l info -S two/ -n two@%h

*  进行测试

# 调用代码
    from workerOne import write_sql_task
    
    write_sql_task.apply_async(args=["is one"])
    # 因为使用的是集群消费， 所以你会看到第一次是第一个worker消费， 第二次是第二个worker消费

#### 广播消费 ####

广播消费就是群发的概念，消息被消费的次数取决于有多少个worker  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70_pic_center 1]

*  worker

from kombu import Queue
    from kombu.common import Broadcast
    
    import logging
    from celery import Celery
    
    # logging
    logging.basicConfig(level=logging.INFO,
                            format='%(asctime)s %(levelname)s %(message)s'
                            )
    logger = logging.getLogger(__name__)
    
    
    # mq config
    app = Celery("node_worker")
    app.conf.task_queues = [
        Broadcast(f'write-mysql-broadcast')
    ]
    app.conf.task_routes = { 
        "write-mysql-broadcast": { 
            'queue': 'write-mysql-broadcast'
       }
    }
    app.conf.broker_url = 'amqp://user:password@127.0.0.1:5672/celeryhost'
    
    # task
    
    @app.task(name=f"write-mysql")
    def write_sql_task(message):
        logger.error(message)

*  启动worker

# 启动worker 1
    celery -A workerOne worker -l info -S one/ -n one@%h
    # 启动worker 2
    celery -A workerOne worker -l info -S two/ -n two@%h

*  进行测试

# 调用代码
    from workerOne import write_sql_task
    
    write_sql_task.apply_async(args=["is one"], queue='write-mysql-broadcast')
    # 因为是广播消费，所以每次发送消息，都会被消费两次

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/f8b5d43c9d9443e9be7fee23c31faf96.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ2ODYzOTk_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/62e7f92b9d5642338f2d343784aa9aa2.png