node与kafka结合

喜欢ヅ旅行 2023-06-26 08:27 26阅读 0赞

### centos7安装kafka ###

> 参考文章：[https://www.cnblogs.com/yangk1996/p/10841588.html][https_www.cnblogs.com_yangk1996_p_10841588.html]

### 安装注意事项（以阿里云服务器为例） ###

1.  在配置config下面的server.properties文件，增加ip映射，如果不增加就是locahost。这样没办法使用java连接host.name=192.168.100.129，这句话去掉，改成如下内容：
2.  advertised.listeners=PLAINTEXT://你的网址:9092，这里我用的是外网ip，代码如下：

#advertised.listeners=PLAINTEXT://your.host.name:9092
    advertised.listeners=PLAINTEXT://IP:9092

1.  将2181,9092设置在安全组里面  
    ![在这里插入图片描述][20191231095134251.png]
2.  剩下的就按照参考文章启动服务即可

### node 代码 ###

从代码我们可以看到 生产者每两秒生产一条数据  
**具体操作流程：**  
1.安装请参考其他文章  
2.启动zk 启动kofka  
3.安装第三方模块依赖 npm install kafka-node  
producer.js

/** * 生产者 */
    
    const kafka = require('kafka-node');
    
    let conn = { 'kafkaHost':'ip:9092'};
    
    var MQ = function (){ 
        this.mq_producers = { };
    }
    
    MQ.prototype.AddProducer = function (conn, handler){ 
        console.log('增加生产者',conn, this);
        let client = new kafka.KafkaClient(conn);
        /** * Producer(client, [options]) * client: 和kafka服务保持连接的client对象 * options：producer的属性 */
        let producer = new kafka.Producer(client);
    
        producer.on('ready', function(){ 
            if(!!handler){ 
                handler(producer);
            }
        });
    
        producer.on('error', function(err){ 
            console.error('producer error ',err.stack);
        });
    
        this.mq_producers['common'] = producer;
        return producer;
    }
    console.log(MQ);
    var mq = new MQ();
    
    mq.AddProducer(conn, function (producer){ 
        producer.createTopics(['broadcast'], function (){ 
            setInterval(function(){ 
                mq.mq_producers['common'].send([{ topic:['broadcast'],
                    messages:[JSON.stringify({ "cmd":"testRpc","value":"Hello World"})]}], function (){ 
                    console.log("..... ");
                })
            }, 2000);
        })
    });

consumer.js

/** * 消费者 */
    
    const kafka = require('kafka-node');
    
    let conn = { 'kafkaHost':'ip:9092'};
    let consumers = [
        { 
            'type': 'consumer',
            'options': { 'autoCommit': true},
            'name':'common',
            'topic':[
                { 'topic': 'broadcast', 'partition': 0}
            ]
        }
    ];
    
    let MQ = function(){ 
    
    }
    
    MQ.prototype.AddConsumer = function (conn, topics, options, handler){ 
        let client = new kafka.KafkaClient(conn);
        let consumer = new kafka.Consumer(client, topics, options);
    
        if(!!handler){ 
            consumer.on('message', handler);
        }
    
        consumer.on('error', function(err){ 
            console.error('consumer error ',err.stack);
        });
    }
    
    var mq = new MQ();
    
    
    mq.AddConsumer(conn, consumers[0].topic, consumers[0].options, function (message){ 
        console.log(message.value);
    });

[https_www.cnblogs.com_yangk1996_p_10841588.html]: https://www.cnblogs.com/yangk1996/p/10841588.html
[20191231095134251.png]: https://img-blog.csdnimg.cn/20191231095134251.png