基于Kafka实现Web应用的一对一与群组消息方案咨询

Hey there! Let's break down how to implement one-to-one and group messaging in your web app using this tech stack. I'll cover everything from setting up Docker Kafka to writing Spring Boot producers/consumers and Angular frontend integration, with practical code snippets along the way.

1. First: Set Up Dockerized Kafka #

First, let's get your Kafka cluster running with Docker. Create a docker-compose.yml file with the following configuration:

version: '3.8'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.4.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.4.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "29092:29092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

Spin up the services with:

docker-compose up -d

2. Spring Boot Setup #

2.1 Add Dependencies #

Add these to your pom.xml (Maven) or equivalent for Gradle:

<dependencies>
    <!-- Spring Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- Spring WebSocket for real-time frontend pushes -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-websocket</artifactId>
    </dependency>
    <!-- JSON Serialization -->
    <dependency>
        <groupId>com.fasterxml.jackson.core</groupId>
        <artifactId>jackson-databind</artifactId>
    </dependency>
</dependencies>

2.2 Configure Kafka in application.properties #

# Producer Config
spring.kafka.producer.bootstrap-servers=localhost:29092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

# Consumer Config
spring.kafka.consumer.bootstrap-servers=localhost:29092
spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.properties.spring.json.trusted.packages=*

3. Implement One-to-One Messaging #

One-to-one messaging targets a specific user. We'll use user-specific topics for simplicity—each user subscribes to their own private topic.

3.1 Message DTO #

Create a ChatMessage class to standardize message data:

public class ChatMessage {
    private String senderId;
    private String recipientId;
    private String content;
    private LocalDateTime timestamp;

    // Getters, Setters, and a constructor for convenience
}

3.2 Private Message Producer #

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class MessageProducer {
    private static final String PRIVATE_TOPIC_PREFIX = "private-messages-";
    private final KafkaTemplate<String, ChatMessage> kafkaTemplate;

    public MessageProducer(KafkaTemplate<String, ChatMessage> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendPrivateMessage(String recipientId, ChatMessage message) {
        String targetTopic = PRIVATE_TOPIC_PREFIX + recipientId;
        kafkaTemplate.send(targetTopic, message);
    }
}

3.3 Private Message Consumer #

Each user subscribes to their own private topic. Use Spring Security to inject the authenticated user's ID dynamically:

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

@Service
public class PrivateMessageConsumer {
    private static final String PRIVATE_TOPIC_PREFIX = "private-messages-";
    private final SimpMessagingTemplate messagingTemplate;

    public PrivateMessageConsumer(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    // Replace ${user.id} with the authenticated user's ID (via Spring Security)
    @KafkaListener(topics = PRIVATE_TOPIC_PREFIX + "${user.id}", groupId = "private-chat-${user.id}")
    public void consumePrivateMessage(ChatMessage message) {
        // Push message to the user's Angular frontend via WebSocket
        messagingTemplate.convertAndSendToUser(
            message.getRecipientId(),
            "/queue/private-messages",
            message
        );
    }
}

4. Implement Group Messaging #

For group chats, we'll use group-specific topics. All members subscribe to the same topic, but each uses a unique consumer group ID so everyone receives every message (same group IDs would split messages between members).

4.1 Group Message Producer #

Extend the existing MessageProducer class:

@Service
public class MessageProducer {
    private static final String GROUP_TOPIC_PREFIX = "group-chat-";
    private final KafkaTemplate<String, ChatMessage> kafkaTemplate;

    // ... existing private message code

    public void sendGroupMessage(String groupId, ChatMessage message) {
        String targetTopic = GROUP_TOPIC_PREFIX + groupId;
        kafkaTemplate.send(targetTopic, message);
    }
}

4.2 Group Message Consumer #

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

@Service
public class GroupMessageConsumer {
    private static final String GROUP_TOPIC_PREFIX = "group-chat-";
    private final SimpMessagingTemplate messagingTemplate;

    public GroupMessageConsumer(SimpMessagingTemplate messagingTemplate) {
        this.messagingTemplate = messagingTemplate;
    }

    // Dynamically subscribe to the group topic when a user joins
    @KafkaListener(topics = "#{groupId}", groupId = "group-chat-user-${user.id}")
    public void consumeGroupMessage(ChatMessage message) {
        // Push message to all group members via WebSocket
        messagingTemplate.convertAndSend(
            "/topic/group-chat-" + message.getGroupId(),
            message
        );
    }
}

Critical Note: Each user's group consumer must have a unique groupId (e.g., include their user ID) to ensure all members receive every message in the group topic.

5. Angular Frontend Integration #

Angular uses WebSocket (via STOMP) to receive real-time messages from Spring Boot.

5.1 WebSocket Service #

Create websocket.service.ts:

import { Injectable } from '@angular/core';
import { Stomp } from '@stomp/stompjs';
import * as SockJS from 'sockjs-client';
import { Observable, Subject } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class WebsocketService {
  private stompClient: any;
  private privateMessageSubject = new Subject<any>();
  private groupMessageSubject = new Subject<any>();

  connect(userId: string) {
    const socket = new SockJS('http://localhost:8080/ws');
    this.stompClient = Stomp.over(socket);

    this.stompClient.connect({}, () => {
      // Subscribe to private messages
      this.stompClient.subscribe(`/user/${userId}/queue/private-messages`, (message: any) => {
        this.privateMessageSubject.next(JSON.parse(message.body));
      });
    });
  }

  joinGroup(groupId: string) {
    this.stompClient.subscribe(`/topic/group-chat-${groupId}`, (message: any) => {
      this.groupMessageSubject.next(JSON.parse(message.body));
    });
  }

  sendPrivateMessage(recipientId: string, message: any) {
    this.stompClient.send('/app/private-message', {}, JSON.stringify({
      recipientId,
      ...message
    }));
  }

  sendGroupMessage(groupId: string, message: any) {
    this.stompClient.send('/app/group-message', {}, JSON.stringify({
      groupId,
      ...message
    }));
  }

  getPrivateMessages(): Observable<any> {
    return this.privateMessageSubject.asObservable();
  }

  getGroupMessages(): Observable<any> {
    return this.groupMessageSubject.asObservable();
  }
}

5.2 Spring Boot WebSocket Config #

Add a WebSocketConfig class to enable STOMP:

import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.simp.config.MessageBrokerRegistry;
import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry config) {
        config.enableSimpleBroker("/topic", "/queue");
        config.setApplicationDestinationPrefixes("/app");
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/ws").withSockJS();
    }
}

6. Key Troubleshooting Tips #

• Docker Network: If Spring Boot runs in a container, ensure it's on the same Docker network as Kafka, and use kafka:9092 as the bootstrap server instead of localhost:29092.
• Consumer Group IDs: For group chats, unique IDs per user are non-negotiable—skip this, and messages will be split between members instead of everyone getting them.
• Authentication: Integrate Spring Security to safely inject user IDs into Kafka listeners and WebSocket subscriptions.

内容的提问来源于stack exchange，提问作者Nafiul Alam Fuji