kafka，redis，生产者消费者，订阅发布

Love The Way You Lie 2021-12-12 01:53 406阅读 0赞

什么是生产者消费者，什么又是订阅发布：

生产消费者模式，指的是由生产者将数据源源不断推送到消息中心，由不同的消费者从消息中心取出数据做自己的处理，在同一类别下，所有消费者拿到的都是同样的数据；

订阅发布模式，本质上也是一种生产消费者模式，不同的是，由订阅者首先向消息中心指定自己对哪些数据感兴趣，发布者推送的数据经过消息中心后，每个订阅者拿到的仅仅是自己感兴趣的一组数据。

这两种模式是使用消息中间件时最常用的，用于功能解耦和分布式系统间的消息通信。

总结成两句话：

生产者消费者：一个消息只能有一个消费者发布者

订阅者：一个消息可以被多个消费者收到

场景应用：

数据接入

假设有一个用户行为采集系统，负责从App端采集用户点击行为数据。通常会将数据上报和数据处理分离开，即App端通过REST API上报数据，后端拿到数据后放入队列中就立刻返回，而数据处理则另外使用Worker从队列中取出数据来做，如下图所示。

![1545259-20190628162651332-269805386.png][]

这样做的好处有：第一，功能分离，上报的API接口不关心数据处理功能，只负责接入数据；第二，数据缓冲，数据上报的速率是不可控的，取决于用户使用频率，采用该模式可以一定程度地缓冲数据；第三，易于扩展，在数据量大时，通过增加数据处理Worker来扩展，提高处理速率。这便是典型的生产消费者模式，数据上报为生产者，数据处理为消费者。(笔者的公司就是用这种方式来实现数据回调)

事件分发  
  假设有一个电商系统，那么，用户“收藏”、“下单”、“付款”等行为都是非常重要的事件，通常后端服务在完成相应的功能处理外，还需要在这些事件点上做很多其他处理动作，比如发送短信通知、记录用户积分等等。我们可以将这些额外的处理动作放到每个模块中，但这并不是优雅的实现，不利于功能解耦和代码维护。

![1545259-20190628162900193-1986497457.png][]  
  我们需要的是一个事件分发系统，在各个功能模块中将对应的事件发布出来，由对其感兴趣的处理者进行处理。这里涉及两个角色：A对B感兴趣，A是处理者，B是事件，由事件处理器完成二者的绑定，并向消息中心订阅事件。服务模块是后端的业务逻辑服务，在不同的事件点发布事件，事件经过消息中心分发给事件处理器对应的处理者。整个流程如下图所示。这边是典型的订阅发布模式。

# kafka生产者
    from pykafka import KafkaClient
    client = KafkaClient(hosts="localhost:9092")  # 可接受多个client
    topic = client.topics[b'topic_demo']  # 选择一个topic
    message = '{"id":"1","value":"3"}'
    # 当有了topic之后呢，可以创建一个producer,来发消息，生产kafka数据,通过字符串形式，
    with topic.get_sync_producer() as producer:
        producer.produce(message.encode("utf-8"))

# kafka消费者
    from kafka import KafkaConsumer
    def callback_process():
        consumer = KafkaConsumer(
            'topic_demo',
            group_id='my-group',
            bootstrap_servers=[
                '{kafka_host}:{kafka_port}'.format(kafka_host='localhost', kafka_port='9092')]
        )
        for message in consumer:
            if message is not None:
                kafka_data = json.loads(message.value.decode("utf-8"))

def run():
        print("start:")
        callback_process()

kafka在win上安装连接：

https://blog.csdn.net/u012050154/article/details/76270655

redis中的订阅与发布，java为发布者，python为订阅者

Redis+jedis在项目中使用以及jedis包下载

https://blog.csdn.net/weixin\_38061311/article/details/86747677

// java
    import redis.clients.jedis.Jedis;
    import java.util.Date;
    public class RedisPub {
        private static final String channel = "cord";
        public static void main(String[] args){
            Jedis jedis = new Jedis("localhost",6379);
            jedis.auth("password");
            String message = new Date().toString();
            jedis.publish(channel, message);
        }
    }

import redis
    
    conn = redis.Redis(host='127.0.0.1', port=6379, password="password")
    # 拿到发布者对象
    pub = conn.pubsub()
    
    pub.subscribe('cord')
    # 可以订阅多个频道
    pub.subscribe('cord1')
    while True:
        msg = pub.parse_response()
        print(msg)
        '''
        [b'subscribe', b'cord', 1]
        [b'subscribe', b'cord1', 2]
        [b'message', b'cord1', b'Fri Jun 28 16:47:27 CST 2019']
        [b'message', b'cord', b'Fri Jun 28 16:47:39 CST 2019']
        '''

参考博文：https://blog.csdn.net/zwgdft/article/details/54633105

转载于:https://www.cnblogs.com/zjbacke/p/11103631.html

[1545259-20190628162651332-269805386.png]: /images/20211211/7c221c1db89c408f9c8b8e3b110d96dd.png
[1545259-20190628162900193-1986497457.png]: /images/20211211/453c59fb9ab24b02b723845330831017.png