Java分布式数据发布者怎么订阅

本文目录导读：

1. 基于Apache Kafka
2. 基于RabbitMQ
3. 基于Redis发布订阅
4. 基于JMS (ActiveMQ)
5. 选择建议

在Java分布式系统中，“发布-订阅”模式通常通过消息中间件实现,以下是几种主流方案及其实现方式：

基于Apache Kafka

生产者（发布者）

import org.apache.kafka.clients.producer.*;
public class KafkaProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", 
            "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", 
            "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(props);
        for (int i = 0; i < 100; i++) {
            producer.send(new ProducerRecord<>("my-topic", 
                Integer.toString(i), "message-" + i));
        }
        producer.close();
    }
}

消费者（订阅者）

import org.apache.kafka.clients.consumer.*;
public class KafkaConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", 
            "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", 
            "org.apache.kafka.common.serialization.StringDeserializer");
        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("my-topic"));
        while (true) {
            ConsumerRecords<String, String> records = 
                consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("offset = %d, key = %s, value = %s%n", 
                    record.offset(), record.key(), record.value());
            }
        }
    }
}

基于RabbitMQ

生产者（发布者）

import com.rabbitmq.client.*;
public class RabbitMQPublisher {
    private final static String EXCHANGE_NAME = "my_exchange";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            String message = "Hello World!";
            channel.basicPublish(EXCHANGE_NAME, "", null, 
                message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

消费者（订阅者）

import com.rabbitmq.client.*;
public class RabbitMQSubscriber {
    private final static String EXCHANGE_NAME = "my_exchange";
    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        System.out.println(" [*] Waiting for messages...");
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println(" [x] Received '" + message + "'");
        };
        channel.basicConsume(queueName, true, deliverCallback, 
            consumerTag -> { });
    }
}

基于Redis发布订阅

配置Redis

<!-- pom.xml -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>

发布者

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
@Component
public class RedisMessagePublisher {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    public void publish(String channel, String message) {
        redisTemplate.convertAndSend(channel, message);
    }
}

订阅者

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;
@Component
public class RedisMessageSubscriber implements MessageListener {
    @Override
    public void onMessage(Message message, byte[] pattern) {
        System.out.println("Received: " + message.toString());
        // 处理消息逻辑
    }
}

配置监听器

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
@Configuration
public class RedisConfig {
    @Bean
    RedisMessageListenerContainer container(
            RedisConnectionFactory connectionFactory,
            MessageListenerAdapter listenerAdapter) {
        RedisMessageListenerContainer container = 
            new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.addMessageListener(listenerAdapter, 
            new ChannelTopic("my-channel"));
        return container;
    }
    @Bean
    MessageListenerAdapter listenerAdapter(
            RedisMessageSubscriber subscriber) {
        return new MessageListenerAdapter(subscriber, "onMessage");
    }
}

基于JMS (ActiveMQ)

生产者

import javax.jms.*;
public class JMSProducer {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new 
            org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616");
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Topic topic = session.createTopic("my-topic");
        MessageProducer producer = session.createProducer(topic);
        TextMessage message = session.createTextMessage("Hello World");
        producer.send(message);
        session.close();
        connection.close();
    }
}

消费者

import javax.jms.*;
public class JMSConsumer implements MessageListener {
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new 
            org.apache.activemq.ActiveMQConnectionFactory("tcp://localhost:61616");
        Connection connection = factory.createConnection();
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        Topic topic = session.createTopic("my-topic");
        MessageConsumer consumer = session.createConsumer(topic);
        consumer.setMessageListener(message -> {
            if (message instanceof TextMessage) {
                try {
                    System.out.println("Received: " + 
                        ((TextMessage) message).getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });
        // 保持连接运行
        Thread.sleep(100000);
        session.close();
        connection.close();
    }
}

选择建议

核心概念对比：

• Kafka：主题分区、消费者组、持久化消息、高吞吐
• RabbitMQ：交换器、队列、路由键、多协议支持
• Redis：频道、轻量级、内存存储、简单模式
• JMS：标准API、Topic/Queue、可靠性高

选择时要考虑吞吐量、持久化需求、消息顺序、事务支持等因素，对于大多数微服务架构,RabbitMQ或Kafka是常见选择。