我如何连接Kafkajs与Socket.io

Question

我一直在学习kafkajs和socket。我对它很陌生，有些事情我似乎不能理解。我已经创建了一个聊天应用程序，基本上通过打开浏览器(客户端)，你可以键入消息，他们得到显示在聊天窗口。我找到了一个教程，让kafka打印"这条消息+ I"。我不想发送到主题并打印消息+ I来打印人们在聊天中输入的内容，我不确定我应该如何做到这一点。

这是我的consumer.js:

const { Kafka } = require("kafkajs")
const clientId = "my-app"
const brokers = ["localhost:9092"]
const topic = "message-log"
const kafka = new Kafka({ clientId, brokers })
// create a new consumer from the kafka client, and set its group ID
// the group ID helps Kafka keep track of the messages that this client
// is yet to receive
const consumer = kafka.consumer({ groupId: clientId })
const consume = async () => {
// first, we wait for the client to connect and subscribe to the given topic
await consumer.connect()
await consumer.subscribe({ topic })
await consumer.run({
// this function is called every time the consumer gets a new message
eachMessage: ({ message }) => {
// here, we just log the message to the standard output
console.log(`received message: ${message.value}`)
},
})
}
module.exports = consume

这是我的producer.js:

// import the `Kafka` instance from the kafkajs library
const { Kafka } = require("kafkajs")
// the client ID lets kafka know who's producing the messages
const clientId = "my-app"
// we can define the list of brokers in the cluster
const brokers = ["localhost:9092"]
// this is the topic to which we want to write messages
const topic = "message-log"
// initialize a new kafka client and initialize a producer from it
const kafka = new Kafka({ clientId, brokers })
const producer = kafka.producer()
// we define an async function that writes a new message each second
const produce = async () => {
await producer.connect()
let i = 0
// after the produce has connected, we start an interval timer
setInterval(async () => {
try {
// send a message to the configured topic with
// the key and value formed from the current value of `i`
await producer.send({
topic,
messages: [
{
key: String(i),
value: "this is message " + i,
},
],
})
// if the message is written successfully, log it and increment `i`
console.log("writes: ", i)
i++
} catch (err) {
console.error("could not write message " + err)
}
}, 1000)
}
module.exports = produce

我知道我应该以某种方式连接主题代理和客户端与套接字。但我不知道怎么做。

这是我的chat.js:

/* Kane connection sto server opos prin
exw tin ikanotita na xrhsimopoihsw to io logo tou library pou phra apo to documentation*/
var socket = io.connect('http://localhost:8000');
// linking Variables toy indexhtml
var message = document.getElementById('message');
var username = document.getElementById('username');
var btn = document.getElementById('send');
var output = document.getElementById('output');
var feedback = document.getElementById('feedback');
// Stelnw events pou ginonte apo ton xristi kai stelnonte ston server
btn.addEventListener('click', function(){
socket.emit('chat', {
message: message.value,
username: username.value
});
message.value = "";
});
message.addEventListener('keypress', function(){
socket.emit('typing', username.value);
})
// Events wste na perimenw to data apo ton server
socket.on('chat', function(data){
feedback.innerHTML = '';
output.innerHTML += '<p><strong>' + data.username + ': </strong>' + data.message + '</p>';
});
socket.on('typing', function(data){
feedback.innerHTML = '<p><em>' + data + ' is typing a message...</em></p>';
});

Answer 1

您需要一个套接字。io服务器。

的例子:

const consume = require('consumer.js');
const produce = require('producer.js');
const { Server } = require('socket.io');
const io = new Server();
consume(({ from, to, message }) => {
io.sockets.emit('newMessage', { from, to, message });
})
io.on('connection', function(socket) {
socket.emit('Hi!', { message: 'Chat connected', id: socket.id });

socket.on('sendMessage', ({ message, to }) => {
produce({ from: socket.id, to, message });
});

});

你还需要修改你的消费者&生产者接受参数和回调。

消费者的例子:

...
const consume = async cb => {
// first, we wait for the client to connect and subscribe to the given topic
await consumer.connect()
await consumer.subscribe({ topic })
await consumer.run({
// this function is called every time the consumer gets a new message
eachMessage: ({ from, to, message }) => {
cb({ from, to, message });
},
});
}

生产商的例子:

const produce = async ({ from, to, message }) => {
producer.send(topic, { from, to, message });
}

别忘了在客户端修改你的chat.js

所有这些都可以优化，只是一个简单的例子