Java 消息代理：企业集成的 5 项基本技术

大家好，这里是架构资源栈！点击上方关注，添加“星标”，一起学习大厂前沿架构！

Java 消息代理通过实现分布式系统之间的可靠通信路径，改变了企业集成。我广泛使用了这些技术，发现它们对于构建可有效扩展的弹性架构至关重要。

Java 企业集成中的消息代理

消息代理充当处理应用程序组件之间的消息验证、路由和传递的中介。它们构成了异步通信的基础，使系统可以在没有直接依赖的情况下进行交互。

核心优势包括服务解耦、提高容错能力以及在流量高峰期间更好地处理负载。根据我的经验，这种架构方法一直在提高系统可靠性。

Apache ActiveMQ

ActiveMQ 是 Java 中最成熟的消息代理之一，提供全面的 JMS 支持以及现代协议。它的成熟度为企业环境带来了稳定性。

我已经在多个生产系统中实现了 ActiveMQ，它的简单配置使得新接触面向消息的中间件的团队也可以使用它。

// Producer example with ActiveMQ
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("order.processing");
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(DeliveryMode.PERSISTENT);TextMessage message = session.createTextMessage("New order: #1001");
message.setStringProperty("OrderType", "Standard");
producer.send(message);// Clean up resources
producer.close();
session.close();
connection.close();

对于消费者来说，实现消息接收需要类似的连接设置：

// Consumer example with ActiveMQ
ConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = factory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
Destination destination = session.createQueue("order.processing");
MessageConsumer consumer = session.createConsumer(destination);consumer.setMessageListener(message -> {if (message instanceof TextMessage) {try {TextMessage textMessage = (TextMessage) message;System.out.println("Received: " + textMessage.getText());System.out.println("Order type: " + textMessage.getStringProperty("OrderType"));} catch (JMSException e) {e.printStackTrace();}}
});// Keep the consumer running (in a real application, you'd have a cleaner approach)
// When ready to shut down:
// consumer.close();
// session.close();
// connection.close();

ActiveMQ 还支持集群以实现高可用性，我已经在生产中对其进行了配置，以确保即使在代理发生故障时也能传递消息。

RabbitMQ

RabbitMQ 在需要复杂路由模式的场景中表现出色。其高级消息队列协议 (AMQP) 的实现通过交换器和队列提供了复杂的消息路由。

我在微服务架构中部署了 RabbitMQ，它的路由功能对于将消息定向到适当的服务非常有用。

// RabbitMQ producer with exchange-based routing
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// Declare exchange and queueString exchangeName = "orders.exchange";String queueName = "orders.processing";String routingKey = "order.created";channel.exchangeDeclare(exchangeName, "topic", true);channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, exchangeName, routingKey);// Create and send messageString message = "{"orderId":"12345","customer":"John Doe","amount":99.95}";channel.basicPublish(exchangeName, routingKey,new AMQP.BasicProperties.Builder().contentType("application/json").deliveryMode(2) // persistent.build(),message.getBytes(StandardCharsets.UTF_8));
}

RabbitMQ 消费者可以实现确认模式来确保可靠的处理：

// RabbitMQ consumer with manual acknowledgment
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();String queueName = "orders.processing";
channel.queueDeclare(queueName, true, false, false, null);
// Limit to processing one message at a time
channel.basicQos(1);DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), StandardCharsets.UTF_8);try {System.out.println("Processing message: " + message);// Process the message...// Acknowledge successful processingchannel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);} catch (Exception e) {// Nack and requeue on failurechannel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);}
};channel.basicConsume(queueName, false, deliverCallback, consumerTag -> {});

RabbitMQ 的可靠性延伸至其管理界面，它提供有关队列性能和系统健康状况的宝贵见解。

阿帕奇·卡夫卡

Kafka 代表了消息传递系统的范式转变，专注于高吞吐量、持久、分布式事件流。它特别适合日志聚合、事件源和实时分析。

我在每天处理数百万事件的系统中实现了 Kafka，它的线性可扩展性至关重要。

// Kafka producer example
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");
props.put("acks", "all");Producer<String, String> producer = new KafkaProducer<>(props);// Send messages with keys for consistent partitioning
for (int i = 1; i <= 5; i++) {String key = "user-" + i;String value = "{"userId":"" + key + "","action":"login","timestamp":" + System.currentTimeMillis() + "}";ProducerRecord<String, String> record = new ProducerRecord<>("user-events", key, value);producer.send(record, (metadata, exception) -> {if (exception == null) {System.out.printf("Message sent to partition %d with offset %d%n", metadata.partition(), metadata.offset());} else {System.err.println("Failed to send message: " + exception.getMessage());}});
}producer.flush();
producer.close();

Kafka的消费者模型采用消费者组进行并行处理：

// Kafka consumer group example
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "user-analytics-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "earliest");
props.put("enable.auto.commit", "false");KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("user-events"));try {while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));for (ConsumerRecord<String, String> record : records) {System.out.printf("Received message: key = %s, value = %s, partition = %d, offset = %d%n",record.key(), record.value(), record.partition(), record.offset());// Process the message...// Manually commit offsetsMap<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();offsetMap.put(new TopicPartition(record.topic(), record.partition()),new OffsetAndMetadata(record.offset() + 1));consumer.commitSync(offsetMap);}}
} finally {consumer.close();
}

Kafka 的优势在于其保留功能以及通过 Kafka Streams 和 KSQL 实现的流处理集成。

Apache ActiveMQ Artemis

Artemis 代表了下一代 ActiveMQ，它建立在高性能核心上，具有更高的吞吐量和集群能力。

我的团队已将旧的 ActiveMQ 部署迁移到 Artemis，并在保持协议兼容性的同时体验到了显著的性能提升。

// Artemis producer using JMS 2.0 API
try (ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616");JMSContext context = cf.createContext(JMSContext.AUTO_ACKNOWLEDGE)) {Queue queue = context.createQueue("orders.priority");JMSProducer producer = context.createProducer();// Set message propertiesproducer.setProperty("priority", "high").setProperty("region", "EMEA").setPriority(8);// Send the messageproducer.send(queue, "Urgent order requiring immediate processing");
}

Artemis 支持 JMS 2.0 简化消费者 API：

// Artemis consumer using JMS 2.0 API with message selector
try (ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory("tcp://localhost:61616");JMSContext context = cf.createContext(JMSContext.AUTO_ACKNOWLEDGE)) {Queue queue = context.createQueue("orders.priority");String selector = "priority = 'high' AND region = 'EMEA'";// Create consumer with selector to filter messagesJMSConsumer consumer = context.createConsumer(queue, selector);// Synchronous receiving with timeoutString message = consumer.receiveBody(String.class, 5000);if (message != null) {System.out.println("Received priority message: " + message);// Process message...}// Or asynchronous with lambda (uncomment to use)/*consumer.setMessageListener(message -> {try {String body = message.getBody(String.class);System.out.println("Received priority message: " + body);// Process message...} catch (JMSException e) {e.printStackTrace();}});*/
}

Artemis 在集群和高可用性方面提供了显著的改进，我利用这一点来构建有弹性的消息传递系统。

Redis 发布/订阅

Redis Pub/Sub 提供了一种轻量级的消息传递解决方案，在需要超低延迟且消息持久性是可选的场景中表现出色。

我使用 Redis Pub/Sub 进行实时通知和系统仪表板，它的简单性和速度非常合适。

// Redis publisher using Jedis
Jedis jedis = new Jedis("localhost");
String channel = "user-notifications";
String message = "{"userId":"user123","message":"New friend request","timestamp":1623412800}";// Publish message to channel
jedis.publish(channel, message);
jedis.close();

订阅 Redis 频道需要专用连接：

// Redis subscriber using Jedis
Jedis jedisSubscriber = new Jedis("localhost");
JedisPubSub pubSub = new JedisPubSub() {@Overridepublic void onMessage(String channel, String message) {System.out.println("Channel: " + channel + " Message: " + message);// Process the message...// To stop listening (from another thread):// this.unsubscribe();}@Overridepublic void onSubscribe(String channel, int subscribedChannels) {System.out.println("Subscribed to: " + channel);}
};// Subscribe to channel(s) - this is a blocking call
jedisSubscriber.subscribe(pubSub, "user-notifications");

虽然 Redis Pub/Sub 不提供消息持久性，但它可以与 Redis Streams 结合使用，以获得更持久的方法：

// Redis Streams example (for persistence)
Jedis jedis = new Jedis("localhost");
String streamKey = "user-activity";
Map<String, String> fields = new HashMap<>();
fields.put("userId", "user123");
fields.put("action", "login");
fields.put("timestamp", String.valueOf(System.currentTimeMillis()));// Add entry to stream
String entryId = jedis.xadd(streamKey, StreamEntryID.NEW_ENTRY, fields);
System.out.println("Added entry with ID: " + entryId);// Read from stream
List<StreamEntry> entries = jedis.xread(XReadParams.xReadParams().count(10).block(2000),Map.of(streamKey, new StreamEntryID("0-0"))
);for (StreamEntry entry : entries) {System.out.println("Entry ID: " + entry.getID());for (Map.Entry<String, String> field : entry.getFields().entrySet()) {System.out.println(field.getKey() + ": " + field.getValue());}
}jedis.close();

Redis 的简单性使其成为轻量级消息传递需求的绝佳选择，或作为更强大的消息代理的补充系统。

集成模式和最佳实践

在所有这些消息代理中，我发现某些集成模式始终很有价值：

消息幂等性确保了消息重新处理的安全，我通过添加唯一的消息 ID 和跟踪已处理的消息来实现这一点：

// Simplified idempotent consumer example
public class IdempotentMessageProcessor {private final Set<String> processedMessageIds = ConcurrentHashMap.newKeySet();public void processMessage(String messageId, String content) {if (processedMessageIds.contains(messageId)) {System.out.println("Message " + messageId + " already processed, skipping");return;}try {// Process the messageSystem.out.println("Processing message: " + content);// Mark as processedprocessedMessageIds.add(messageId);} catch (Exception e) {System.err.println("Failed to process message: " + e.getMessage());// Don't mark as processed to allow retry}}
}

死信队列（DLQ）可以妥善处理消息处理失败：

// Example of configuring a dead letter queue in ActiveMQ
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");
Connection connection = connectionFactory.createConnection();
connection.start();
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);// Create the original destination
Queue originalQueue = session.createQueue("orders.processing");// Create the DLQ
Queue dlq = session.createQueue("ActiveMQ.DLQ.orders.processing");// Configure a producer to use the original queue
MessageProducer producer = session.createProducer(originalQueue);// Configure the message to use the DLQ if it can't be delivered
TextMessage message = session.createTextMessage("Order data that might fail processing");
message.setIntProperty("JMSXDeliveryCount", 0);
message.setStringProperty("JMS_AMQP_FirstAcquirer", "true");// Send the message
producer.send(message);

当下游服务出现故障时，断路器可防止系统过载：

// Simple circuit breaker pattern for messaging
public class MessagingCircuitBreaker {private enum State { CLOSED, OPEN, HALF_OPEN }private State state = State.CLOSED;private int failureCount = 0;private final int failureThreshold;private long openTimestamp;private final long resetTimeoutMs;public MessagingCircuitBreaker(int failureThreshold, long resetTimeoutMs) {this.failureThreshold = failureThreshold;this.resetTimeoutMs = resetTimeoutMs;}public synchronized boolean allowRequest() {if (state == State.CLOSED) {return true;} else if (state == State.OPEN) {// Check if timeout has elapsed to transition to half-openif (System.currentTimeMillis() - openTimestamp > resetTimeoutMs) {state = State.HALF_OPEN;return true;}return false;} else { // HALF_OPENreturn true;}}public synchronized void recordSuccess() {if (state == State.HALF_OPEN) {state = State.CLOSED;failureCount = 0;}}public synchronized void recordFailure() {failureCount++;if (state == State.CLOSED && failureCount >= failureThreshold) {state = State.OPEN;openTimestamp = System.currentTimeMillis();} else if (state == State.HALF_OPEN) {state = State.OPEN;openTimestamp = System.currentTimeMillis();}}
}

结论

选择正确的消息代理取决于特定要求：

• ActiveMQ 提供稳定性和丰富的功能集，适合传统企业集成。
• RabbitMQ 擅长微服务架构所需的复杂消息路由模式。
• Kafka 为数据密集型应用程序提供高吞吐量流处理能力。
• Artemis 为 JMS 应用程序提供了下一代性能。
• Redis Pub/Sub 提供轻量级消息传递，用于实时更新。

根据我在各个行业实施这些系统的经验，最成功的集成将适当的技术选择与可靠的架构模式相结合。消息代理不仅可以连接系统，还可以实现可随时间推移而发展的弹性、可扩展的架构。

消息代理的采用不断提高我所研究的分布式系统的可靠性和可维护性，与传统的单片方法相比，可以实现更细粒度的扩展和更好的故障隔离。

---

原文地址：https://mp.weixin.qq.com/s/9sQ0kQVPBXP0OzVVbd8nQA