目录

项目准备

配置pom.xml

配置application.yml

创建config、service类

简单的消息发送与接收

消息模型

p2p（Queue消息模型）

pub/sub（Topic消息模型）

进阶API

五种消息类型

TextMessage

MapMessage

BytesMessage

StreamMessage

ObjectMessage

QosSettings使用

延迟发送消息

死信队列

自定义死信队列名称

非持久化的消息也进入死信队列

过期消息不进死信队列

项目准备

准备好一个SpringBoot项目

配置pom.xml

在pom.xml的dependencies标签中添加以下依赖包

    org.springframework.boot
    spring-boot-starter-activemq




    org.springframework.boot
    spring-boot-starter-test
    test

配置application.yml

spring:
  activemq:
    # mq的链接地址
    broker-url: tcp://localhost:61616
    # 用户名
    user: admin
    # 密码
    password: admin
  jms:
    # 消息模型 true：发布/订阅消息模型  false：点对点消息模型
    pub-sub-domain: true

创建config、service类

首先创建config包，并创建名为ActiveMqConfig的类

@Configuration
@EnableJms
public class ActiveMqConfig {

    
    @Bean
    public JmsListenerContainerFactory jmsListenerContainerTopic(ConnectionFactory activeMQConnectionFactory) {
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        bean.setPubSubDomain(true);
        bean.setConnectionFactory(activeMQConnectionFactory);
        return bean;
    }

    
    @Bean
    public JmsListenerContainerFactory jmsListenerContainerQueue(ConnectionFactory activeMQConnectionFactory) {
        DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
        bean.setConnectionFactory(activeMQConnectionFactory);
        return bean;
    }
}

创建service包，并创建名为ReceiverService的类

@Service
public class ReceiverService {

    
    @JmsListener(destination = "topicMsg", containerFactory = "jmsListenerContainerTopic")
    public void topicMsg(String msg) {
        System.out.println("发布/订阅消息：" + msg);
    }
    
    
    @JmsListener(destination = "queueMsg", containerFactory = "jmsListenerContainerQueue")
    public void queueMsg(String msg) {
        System.out.println("点对点消息：" + msg);
    }

}

再创建一个消息发送类

@Service
public class SendService {

    
}

此时的项目结构如下图

简单的消息发送与接收

在SendService中引入JmsMessagingTemplate，创建一个发送方法，代码如下

@Autowired
private JmsMessagingTemplate jmsMessagingTemplate;

public void send(String destination, String msg) {
    jmsMessagingTemplate.convertAndSend(destination, msg);
}

在ActiveMqDemoApplicationTests类中引入SendService，创建一个测试方法，调用send方法

@Autowired
private SendService sendService;

private String topicMsg = "topicMsg";

private String queueMsg = "queueMsg";

@Test
void send() {
    sendService.send(topicMsg, "msg");
}

运行之后消息就被打印出来了

 可以通过切换destination来测试不同的消息模型，使用queueMsg作为destination时需要将配置文件中的pub-sub-domain设置为false

消息模型

p2p（Queue消息模型）

基于点对点的方式，消费者从Queue取出消息并消费，被消费的消息将不再储存，允许多个消费者监听一个Queue，但对于一个消息只会存在一个消费者，消息没被消费时，将一直存储

pub/sub（Topic消息模型）

 发布/订阅的方式，生产者将消息发布到Topic中，监听该Topic的所有消费者将获取到消息，消息发布时，如果没有消费者也不会保存消息

进阶API

JmsMessagingTemplate对JmsTemplate进行了封装，使用JmsTemplate可以自定义消息发送方法

五种消息类型

自定义消息类型有TextMessage、MapMessage、BytesMessage、StreamMessage、ObjectMessage，发送不同类型的消息，消费者也应设置对应的消息接收类型

// 引入JmsTemplate
@Autowired
private JmsTemplate jmsTemplate;

TextMessage

jmsTemplate.send(destination, session -> session.createTextMessage("msg"));

MapMessage

jmsTemplate.send(destination, session -> {
    MapMessage mapMessage = session.createMapMessage();
    mapMessage.setIntProperty("int", 1);
    mapMessage.setStringProperty("a", "b");
    mapMessage.setString("key","value");
    mapMessage.setBoolean("boolean",false);
    mapMessage.setInt("int", 1);
    return mapMessage;
});

BytesMessage

jmsTemplate.send(destination, session -> {
    BytesMessage bytesMessage = session.createBytesMessage();
    // 发送字节
    bytesMessage.writeBytes("bytes".getBytes());
    // 发送文本
    bytesMessage.writeUTF("文本");
    return bytesMessage;
});

StreamMessage

jmsTemplate.send(destination, session -> {
    StreamMessage streamMessage = session.createStreamMessage();
    streamMessage.writeBoolean(true);
    streamMessage.writeByte((byte) 10);
    streamMessage.writeBytes("text".getBytes());
    streamMessage.writeChar('A');
    streamMessage.writeDouble(55.3D);
    streamMessage.writeFloat(79.1F);
    streamMessage.writeInt(37);
    streamMessage.writeLong(56652L);
    streamMessage.writeObject("text");
    streamMessage.writeShort((short) 333);
    streamMessage.writeString("text");
    return streamMessage;
});

ObjectMessage

jmsTemplate.send(destination, session -> {
    ObjectMessage objectMessage = session.createObjectMessage();
    objectMessage.setObject("ABC");
    return objectMessage;
});

QosSettings使用

QosSettings可以设置消息优先级、存活时间、消息是否持久化

要支持设置消息优先级需要先在activemq.xml的配置文件中policyEntries标签内添加以下配置，queue就是设置优先级的队列名称，优先级分为0-9，默认是4，9最高

public void sendMsg(String destination, Integer priority) {
    QosSettings settings = new QosSettings();
    // 设置优先级
    settings.setPriority(priority);
    // 设置发送模式 NON_PERSISTENT 不持久化， PERSISTENT 持久化
    settings.setDeliveryMode(DeliveryMode.PERSISTENT);
    // 设置消息超时时间，超时的消息无法被消费，要消费可以从死信队列中重发
    settings.setTimeToLive(5000);
    // 将设置传入
    JmsTemplate jmsTemplate = jmsMessagingTemplate.getJmsTemplate();
    jmsTemplate.setQosSettings(settings);
    jmsMessagingTemplate.convertAndSend(destination,"优先级为" + priority);
}

@Test
void sendMsg() {
    Random random = new Random(10);
    for(int i = 0; i < 100; i++) {
        int priority = random.nextInt(10);
        sendService.sendMsg(queueMsg, priority);
    }
}

延迟发送消息

要支持延迟发送消息，需要在activemq.xml配置文件的broker标签中添加schedulerSupport="true"，如：

代码示例：

public void sendMsg(String destination, String msg) {
    Map headers = new HashMap<>();
    // 发送延迟队列，延迟10秒,单位毫秒，支持Cron表达式定时发送
    headers.put(ScheduledMessage.AMQ_SCHEDULED_DELAY, 10000);
    jmsMessagingTemplate.convertAndSend(destination,msg, headers);
}

死信队列

死信队列 (Dead Letter Queue) 用来保存处理失败或者过期的消息，这些消息通过创建死信队列对应名称的消费者来获取

自定义死信队列名称

通常情况为了易于维护会让自定义死信队列名称，且每个队列不共用一个死信队列，在activemq.xml文件中可以修改死信队列名称，如以下示例，将queueMsg队列的死信队列名称定义为以DLQ为前缀

topic和queue都可以保存死信，通过useQueueForQueueMessages来配置，false为topic，true为queue，默认是true

非持久化的消息也进入死信队列

默认情况下只有持久化的消息才会进入死信队列，要将非持久化的消息进入死信队列需要更改配置，如：

其中processNonPersistent表示是否非持久化的消息进入死信队列，默认为false

过期消息不进死信队列

有些业务场景，过期的消息不需要处理，没必要进入死信队列

其中processExpired="false"就是让过期消息不进入死信队列