Java后端-JavaEE-JMS消息服务与异步通信机制-6

1
2
3
4
5
6
7

┌─────────────┐     ┌─────────────────────────────────────┐     ┌─────────────┐
│  生产者      │     │          JMS 提供者（Broker）        │     │  消费者      │
│  (Producer) │────▶│  ┌─────────┐    ┌───────────────┐  │────▶│  (Consumer) │
│             │     │  │ Queue   │    │    Topic      │  │     │             │
│             │     │  │(点对点) │    │  (发布/订阅)   │  │     │             │
└─────────────┘     │  └─────────┘    └───────────────┘  │     └─────────────┘
                    └─────────────────────────────────────┘

1
2
3
4
5
6
7

// 同步接收（阻塞）
Message message = consumer.receive(); // 一直等待
// 或设置超时时间
Message message = consumer.receive(5000); // 最多等待5秒
if (message != null) {
    // 处理消息
}

1
2
3
4
5
6
7

consumer.setMessageListener(new MessageListener() {
    @Override
    public void onMessage(Message message) {
        // 消息到达时自动调用
        // 处理消息...
    }
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
// 创建生产者、发送消息...
try {
    // 发送多条消息
    producer.send(message1);
    producer.send(message2);
    // 提交事务——两条消息同时成功
    session.commit();
} catch (Exception e) {
    // 回滚事务——两条消息都不发送
    session.rollback();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// 使用 JTA 分布式事务（伪代码示意）
@Transactional  // 容器管理分布式事务
public void createOrder(Order order) {
    // 1. 写入数据库（参与分布式事务）
    entityManager.persist(order);
    
    // 2. 发送 JMS 消息（参与同一个分布式事务）
    jmsProducer.send(orderQueue, orderMessage);
    
    // 方法正常结束 → 两阶段提交，数据库和消息同时提交
    // 方法抛出异常 → 两阶段回滚，数据库和消息同时回滚
}

1
2
3
4

// 设置消息持久化
producer.setDeliveryMode(DeliveryMode.PERSISTENT);
// 或单独设置每条消息
message.setJMSDeliveryMode(DeliveryMode.PERSISTENT);

1
2

// 设置消息 60 秒后过期
producer.setTimeToLive(60000);

1

producer.setPriority(9);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

┌────────────┐     ┌──────────────────────────────────────────────────────┐     ┌────────────┐
│   用户      │     │                    订单系统                         │     │   库存系统   │
│  下单       │────▶│  ┌─────────────┐    ┌──────────────────────────┐  │     │            │
└────────────┘     │  │ 订单服务     │───▶│  Queue: order.created     │  │────▶│ 扣减库存    │
                   │  │ (保存订单)   │    └──────────────────────────┘  │     │            │
                   │  └─────────────┘                                   │     └──────┬─────┘
                   │         ▲                                          │            │
                   │         │                                          │            │
                   │  ┌──────┴─────┐    ┌──────────────────────────┐  │            │
                   │  │ 订单服务   │◀───│  Queue: inventory.updated │  │◀───────────┘
                   │  │(更新状态)  │    └──────────────────────────┘  │
                   │  └────────────┘                                   │
                   └──────────────────────────────────────────────────────┘

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

<!-- Jakarta EE 9+ 使用 jakarta.jms -->
<dependency>
    <groupId>jakarta.jms</groupId>
    <artifactId>jakarta.jms-api</artifactId>
    <version>3.1.0</version>
    <scope>provided</scope>
</dependency>

<!-- ActiveMQ Artemis 客户端（JMS 实现） -->
<dependency>
    <groupId>org.apache.activemq</groupId>
    <artifactId>artemis-jms-client</artifactId>
    <version>2.31.2</version>
</dependency>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

import jakarta.ejb.Stateless;
import jakarta.ejb.TransactionAttribute;
import jakarta.ejb.TransactionAttributeType;
import jakarta.jms.*;
import jakarta.annotation.Resource;
import jakarta.persistence.EntityManager;
import jakarta.persistence.PersistenceContext;

@Stateless
public class OrderService {
    
    @PersistenceContext
    private EntityManager em;
    
    @Resource(lookup = "java:jboss/queue/orderQueue")
    private Queue orderQueue;
    
    @Resource(lookup = "java:jboss/ConnectionFactory")
    private ConnectionFactory connectionFactory;
    
    @TransactionAttribute(TransactionAttributeType.REQUIRED)
    public void createOrder(Order order) throws Exception {
        // 1. 保存订单到数据库
        em.persist(order);
        
        // 2. 发送 JMS 消息通知库存系统
        try (JMSContext context = connectionFactory.createContext(
                JMSContext.AUTO_ACKNOWLEDGE)) {
            JMSProducer producer = context.createProducer();
            // 设置持久化
            producer.setDeliveryMode(DeliveryMode.PERSISTENT);
            
            // 创建消息（使用 TextMessage 传递 JSON）
            TextMessage message = context.createTextMessage();
            String orderJson = String.format(
                "{\"orderId\":%d,\"productId\":%d,\"quantity\":%d}",
                order.getId(), order.getProductId(), order.getQuantity()
            );
            message.setText(orderJson);
            
            // 发送消息
            producer.send(orderQueue, message);
            System.out.println("订单消息已发送，订单ID：" + order.getId());
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

import jakarta.ejb.MessageDriven;
import jakarta.jms.Message;
import jakarta.jms.MessageListener;
import jakarta.jms.TextMessage;
import jakarta.jms.JMSException;
import jakarta.ejb.ActivationConfigProperty;
import jakarta.persistence.EntityManager;
import jakarta.persistence.PersistenceContext;
import jakarta.ejb.EJB;

@MessageDriven(
    activationConfig = {
        @ActivationConfigProperty(
            propertyName = "destinationType",
            propertyValue = "jakarta.jms.Queue"
        ),
        @ActivationConfigProperty(
            propertyName = "destination",
            propertyValue = "queue/orderQueue"
        ),
        @ActivationConfigProperty(
            propertyName = "acknowledgeMode",
            propertyValue = "Auto-acknowledge"
        )
    }
)
public class InventoryConsumerMDB implements MessageListener {
    
    @PersistenceContext
    private EntityManager em;
    
    @EJB
    private InventoryService inventoryService;
    
    @EJB
    private OrderStatusNotifier notifier;  // 发送状态更新消息
    
    @Override
    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                TextMessage textMsg = (TextMessage) message;
                String orderJson = textMsg.getText();
                System.out.println("库存服务收到订单消息：" + orderJson);
                
                // 解析 JSON，扣减库存
                // 假设解析得到 orderId, productId, quantity
                // inventoryService.deductStock(productId, quantity);
                
                // 扣减成功后，发送状态更新消息
                notifier.sendOrderUpdated(orderId, "INVENTORY_DEDUCTED");
                
                System.out.println("库存扣减完成，订单ID：" + orderId);
            }
        } catch (JMSException e) {
            System.err.println("处理订单消息失败：" + e.getMessage());
            // 抛出 RuntimeException 触发消息重发
            throw new RuntimeException("库存扣减失败", e);
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import jakarta.ejb.Stateless;
import jakarta.jms.*;
import jakarta.annotation.Resource;

@Stateless
public class OrderStatusNotifier {
    
    @Resource(lookup = "java:jboss/queue/orderStatusQueue")
    private Queue statusQueue;
    
    @Resource(lookup = "java:jboss/ConnectionFactory")
    private ConnectionFactory connectionFactory;
    
    public void sendOrderUpdated(Long orderId, String status) {
        try (JMSContext context = connectionFactory.createContext(
                JMSContext.AUTO_ACKNOWLEDGE)) {
            TextMessage message = context.createTextMessage();
            message.setText(String.format(
                "{\"orderId\":%d,\"status\":\"%s\"}",
                orderId, status
            ));
            context.createProducer().send(statusQueue, message);
            System.out.println("订单状态更新消息已发送：" + orderId + " -> " + status);
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

@MessageDriven(
    activationConfig = {
        @ActivationConfigProperty(
            propertyName = "destinationType",
            propertyValue = "jakarta.jms.Queue"
        ),
        @ActivationConfigProperty(
            propertyName = "destination",
            propertyValue = "queue/orderStatusQueue"
        )
    }
)
public class OrderStatusConsumerMDB implements MessageListener {
    
    @PersistenceContext
    private EntityManager em;
    
    @Override
    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                TextMessage textMsg = (TextMessage) message;
                String json = textMsg.getText();
                // 解析 JSON，更新订单状态
                // order.setStatus(status);
                // em.merge(order);
                System.out.println("订单状态已更新：" + json);
            }
        } catch (JMSException e) {
            throw new RuntimeException("更新订单状态失败", e);
        }
    }
}