KafkaJS如何在远端网站/服务器运行消费者以实现实时数据展示？

EventSource要求服务端返回的响应MIME类型必须是text/event-stream，你直接把KafkaJS消费者脚本script.js作为EventSource的请求地址是错误的，浏览器会把它当成普通JS文件返回application/javascript类型，自然不符合规范要求。另外KafkaJS是Node.js端运行的库，无法直接在浏览器环境下执行，你之前编写的消费者脚本本身就只能跑在服务端，不能直接被前端调用。

• 搭建中间层服务：你需要写一个简单的Node.js服务（可以用Express、Koa这类框架），这个服务同时承担两个职责：第一是运行你已经写好的KafkaJS消费者，持续接收Kafka推送的消息；第二是提供专门的SSE接口，接口响应头必须设置Content-Type: text/event-stream、Cache-Control: no-cache、Connection: keep-alive，如果前后端端口不同还要额外配置跨域响应头。
• 前端代码调整：把EventSource的请求地址改成你自己实现的SSE接口地址，不要直接指向script.js，原来的消息解析、绘图逻辑可以保留复用。

服务端（Node.js + Express）代码 #

const express = require('express');
const { Kafka } = require('kafkajs');
const app = express();
const port = 3000;

// 跨域配置，根据实际业务场景调整允许的域名
app.use((req, res, next) => {
  res.header('Access-Control-Allow-Origin', '*');
  res.header('Access-Control-Allow-Methods', 'GET,PUT,POST,DELETE');
  res.header('Access-Control-Allow-Headers', 'Content-Type');
  next();
});

// SSE接口实现
app.get('/kafka-stream', (req, res) => {
  // 设置SSE必需的响应头
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive'
  });

  // Kafka消费者逻辑，替换成你自己的Kafka配置和主题信息
  const kafka = new Kafka({
    clientId: 'web-consumer',
    brokers: ['你的Kafka broker地址']
  });
  const consumer = kafka.consumer({ groupId: 'web-consumer-group' });

  const runConsumer = async () => {
    await consumer.connect();
    await consumer.subscribe({ topic: '你的Kafka消息主题', fromBeginning: false });
    await consumer.run({
      eachMessage: async ({ topic, partition, message }) => {
        // SSE格式要求：每条消息以data:开头，两个换行符结尾
        res.write(`data: ${message.value.toString()}\n\n`);
      },
    });
  };
  runConsumer().catch(console.error);

  // 前端断开连接时销毁消费者释放资源
  req.on('close', () => {
    consumer.disconnect();
    res.end();
  });
});

app.listen(port, () => {
  console.log(`SSE服务运行端口: ${port}`);
});

前端代码 #

// 替换成你自己的SSE接口地址
var source = new EventSource('http://localhost:3000/kafka-stream');
source.addEventListener('message', function(e){
  console.log('收到消息');
  const obj = JSON.parse(e.data);
  console.log(obj);
  // 在此处添加你的绘图逻辑即可
});
// 增加错误处理避免异常
source.addEventListener('error', function(e) {
  console.error('SSE连接异常', e);
  source.close();
});

• KafkaJS仅支持Node.js环境运行，不能直接引入前端HTML文件中执行
• SSE接口返回的内容必须严格遵循格式要求：每条消息以data: 开头，以两个换行符\n\n结束
• 如果站点访问量较大，可以考虑在中间层加消息缓存或者用WebSocket替代SSE，避免同时维护大量长连接消耗服务端资源
• Kafka的消费者组ID要设置合理，避免多个前端请求重复消费消息或者漏消费

内容的提问来源于stack exchange，提问作者TechnoTerry