2024年3月10日
知识积累
为什么使用Kafka?
不使用Rabbitmq或者Rocketmq是因为Kafka是Hadoop集群下的组成部分,对于大数据的相关开发适应性好,且当前业务场景下不需要使用死信队列,不过要注意Kafka对于更新时间慢的数据拉取也较慢,因此对与实时性要求高可以选择其他MQ。
使用消息队列是因为该中间件具有实时性,且可以作为广播进行消息分发。
为什么使用SSE?
使用Websocket传输信息的时候,会转成二进制数据,产生一定的时间损耗,SSE直接传输文本,不存在这个问题
由于Websocket是双向的,读取日志的时候,如果有人连接ws日志,会发送大量异常信息,会给使用段和日志段造成问题;SSE是单向的,不需要考虑这个问题,提高了安全性
另外就是SSE支持断线重连;Websocket协议本身并没有提供心跳机制,所以长时间没有数据发送时,会将这个连接断掉,因此需要手写心跳机制进行实现。
此外,由于是长连接的一个实现方式,所以SSE也可以替代Websocket实现扫码登陆(比如通过SSE的超时组件在实现二维码的超时功能,具体实现我可以整理一下)
另外,如果是普通项目,不需要过高的实时性,则不需要使用Websocket,使用SSE即可
代码实现
Java代码
pom.xml引入SSE和Kafka
<!-- SSE,一般springboot开发web应用的都有 --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency>
<!-- kafka,最主要的是第一个,剩下两个是测试用的 --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency<groupId>org.springframework.kafka</groupId><artifactId>spring-kafka-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.4.0</version></dependency>
application.properties增加Kafka配置信息
# KafkaProperties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=community-consumer-group
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
配置Kafka信息
@Configuration
public class KafkaProducerConfig {@Value("${spring.kafka.bootstrap-servers}")private String bootstrapServers;@Beanpublic Map<String, Object> producerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);return props;}@Beanpublic ProducerFactory<String, String> producerFactory() {return new DefaultKafkaProducerFactory<>(producerConfigs());}@Beanpublic KafkaTemplate<String, String> kafkaTemplate() {return new KafkaTemplate<>(producerFactory());}}
配置controller,通过web方式开启效果
@RestController
@RequestMapping(path = "sse")
public class KafkaSSEController {private static final Map<String, SseEmitter> sseCache = new ConcurrentHashMap<>();@Resourceprivate KafkaTemplate<String, String> kafkaTemplate;/*** @param message* @apiNote 发送信息到Kafka主题中*/@PostMapping("/send")public void sendMessage(@RequestBody String message) {kafkaTemplate.send("my-topic", message);}/*** 监听Kafka数据** @param message*/@KafkaListener(topics = "my-topic", groupId = "my-group-id")public void consume(String message) {System.out.println("Received message: " + message);}/*** 连接sse服务** @param id* @return* @throws IOException*/@GetMapping(path = "subscribe", produces = {MediaType.TEXT_EVENT_STREAM_VALUE})public SseEmitter push(@RequestParam("id") String id) throws IOException {// 超时时间设置为5分钟,用于演示客户端自动重连SseEmitter sseEmitter = new SseEmitter(5_60_000L);// 设置前端的重试时间为1s// send(): 发送数据,如果传入的是一个非SseEventBuilder对象,那么传递参数会被封装到 data 中sseEmitter.send(SseEmitter.event().reconnectTime(1000).data("连接成功"));sseCache.put(id, sseEmitter);System.out.println("add " + id);sseEmitter.send("你好", MediaType.APPLICATION_JSON);SseEmitter.SseEventBuilder data = SseEmitter.event().name("finish").id("6666").data("哈哈");sseEmitter.send(data);// onTimeout(): 超时回调触发sseEmitter.onTimeout(() -> {System.out.println(id + "超时");sseCache.remove(id);});// onCompletion(): 结束之后的回调触发sseEmitter.onCompletion(() -> System.out.println("完成!!!"));return sseEmitter;}/*** http://127.0.0.1:8080/sse/push?id=7777&content=%E4%BD%A0%E5%93%88aaaaaa* @param id* @param content* @return* @throws IOException*/@ResponseBody@GetMapping(path = "push")public String push(String id, String content) throws IOException {SseEmitter sseEmitter = sseCache.get(id);if (sseEmitter != null) {sseEmitter.send(content);}return "over";}@ResponseBody@GetMapping(path = "over")public String over(String id) {SseEmitter sseEmitter = sseCache.get(id);if (sseEmitter != null) {// complete(): 表示执行完毕,会断开连接sseEmitter.complete();sseCache.remove(id);}return "over";}}
前端方式
<html><head><script>console.log('start')const clientId = "your_client_id_x"; // 设置客户端IDconst eventSource = new EventSource(`http://localhost:9999/v1/sse/subscribe/${clientId}`); // 订阅服务器端的SSEeventSource.onmessage = event => {console.log(event.data)const message = JSON.parse(event.data);console.log(`Received message from server: ${message}`);};// 发送消息给服务器端 可通过 postman 调用,所以下面 sendMessage() 调用被注释掉了function sendMessage() {const message = "hello sse";fetch(`http://localhost:9999/v1/sse/publish/${clientId}`, {method: "POST",headers: { "Content-Type": "application/json" },body: JSON.stringify(message)});console.log('dddd'+JSON.stringify(message))}// sendMessage()</script></head>
</html>