kafka-node+socket.io 测试配置
2024-09-04 12:07:01
1.安装需要插件
npm install express
npm install socket.io
npm install kafka-node
2.kafkatest.js文件
var express = require('express');
var app = express(); var server = require('http').createServer(app);
var io = require('socket.io')(server);
var kafka = require('kafka-node');
var users = [];
app.use('/', express.static(__dirname + '/')); app.get('/send', function (req, res) {
var msg=req.query.msg;
var Producer = kafka.Producer,
client = new kafka.Client('localhost:2181'),
producer = new Producer(client);
payloads = [
{ topic: 'test', messages: msg, partition: 0 },
];
producer.on('ready', function(){
producer.send(payloads, function(err, data){
console.log(data);
// socket.emit('server_counter',data);
});
});
producer.on('error', function(err){}) res.send('输入消息='+msg);
})
server.listen(8080); setTimeout(function(){
var Consumer = kafka.Consumer;
var Offset = kafka.Offset;
var topic = 'test'; var client = new kafka.Client('localhost:2181');//'localhost:2181'
var topics = [{ topic: topic, partition: 0 }];
var options = { autoCommit: false };//, fetchMaxWaitMs: 1000, fetchMaxBytes: 1024 * 1024 var consumer = new Consumer(client, topics, options);
var offset = new Offset(client);
consumer.on('message', function (message) {
console.log(message);
io.sockets.emit('server_counter',message);
}); consumer.on('error', function (err) {
console.log('error', err);
});
consumer.on('offsetOutOfRange', function (topic) {
topic.maxNum = 2;
offset.fetch([topic], function (err, offsets) {
if (err) {
return console.error(err);
}
var min = Math.min.apply(null, offsets[topic.topic][topic.partition]);
consumer.setOffset(topic.topic, topic.partition, min);
});
}); },1000);
3.counter.html
<!DOCTYPE html>
<html> <head>
<title>socket</title>
</head> <body> <div style="margin: 0 auto;" id='msg'> </div>
</body>
<script type="text/javascript" src="./node_modules/socket.io-client/dist/socket.io.js"></script>
<script type="text/javascript">
var socket=io.connect('localhost:8080'),//与服务器进行连接
send=document.getElementById('sendBtn'),
leave=document.getElementById('leaveBtn'); //接收来自服务端的信息事件
socket.on('server_counter',function(msg){
var div =document.createElement('div');
div.innerHTML=JSON.stringify(msg);
document.getElementById('msg').appendChild(div);
})
</script> </html>
3.效果展示
启动 kafkatest.js
打开http://localhost:8080/counter.html
打开http://localhost:8080/send?msg=谢大神你好
最新文章
- React-Native android在windows下的踩坑记
- JMS开发(一):基础理论认知
- MySQL索引视图
- C#解析JSON数据
- ssh伪登陆执行远程主机脚本命令 C程序基于ssh passwordless执行远程主机命令及基于配置文件的验证伪登陆执行命令
- Mysql ODBC 5.1 Driver免安装脚本
- linux 下一个 jira-6.3.6 组态 皴 翻译 迁移数据库
- ImageView.ScaleType设置图解
- 详细了解 int? 类型
- monkeyrunner简介
- GitHub下载安装以及开源项目
- Caffe上用SSD训练和测试自己的数据
- [Android] SeekBar---可拖动进度条
- Java继承访问权限
- Python读取ini配置文件封装方法
- node中__dirname、__filename、process.cwd()表示的路径
- 【PaPaPa】系统架构搭建浅析 - 人人可以搭架构
- [转帖] .net 2.1 是 LTS长期支持版本.
- 样式 bootstrap purecss Amaze UI 推荐
- acdream 1735 输油管道 贪心