Java开发中常用的消息队列工具 ActiveMQ

JAVA_HOME="/opt/jdk1.8.0_152"

JAVA_CMD="/opt/jdk1.8.0_152/bin"

ln -s  /opt/activemq/bin/activemq  /etc/init.d/activemq（软连接必须使用绝对路径）

chkconfig --add activemq

# 禁止使用

# cp /opt/activemq/bin/activemq /etc/init.d/activemq

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-activemq</artifactId>
   <exclusions>
      <exclusion>
         <groupId>org.slf4j</groupId>
         <artifactId>slf4j-log4j12</artifactId>
      </exclusion>
   </exclusions>
</dependency>

<dependency>
   <groupId>org.apache.activemq</groupId>
   <artifactId>activemq-pool</artifactId>
   <version>5.15.2</version>
   <exclusions>
      <exclusion>
         <groupId>org.slf4j</groupId>
         <artifactId>slf4j-log4j12</artifactId>
      </exclusion>
   </exclusions>
</dependency>

public class ProducerTest {
    public static void main(String[] args) throws JMSException {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.67.201:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        // 创建session 第一个参数表示是否支持事务，false时，第二个参数Session.AUTO_ACKNOWLEDGE自动签收，Session.CLIENT_ACKNOWLEDGE手动签收，DUPS_OK_ACKNOWLEDGE 订阅时签收其中一个
        // 第一个参数设置为true时，第二个参数可以忽略 服务器设置为SESSION_TRANSACTED
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建队列
        Queue queue = session.createQueue("Atguigu");

        MessageProducer producer = session.createProducer(queue);
        // 创建消息对象
        ActiveMQTextMessage activeMQTextMessage = new ActiveMQTextMessage();
        activeMQTextMessage.setText("hello ActiveMq!");
        // 发送消息
        producer.send(activeMQTextMessage);
        producer.close();
        connection.close();
    }
}

public class ConsumerTest {
    public static void main(String[] args) throws JMSException {
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, "tcp://192.168.67.201:61616");
        // 创建连接
        Connection connection = activeMQConnectionFactory.createConnection();
        connection.start();
        // 创建会话
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建队列
        Queue queue = session.createQueue("Atguigu");
        // 创建Consumer
        MessageConsumer consumer = session.createConsumer(queue);
        // 接收消息
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                // 参数就是收到的消息
                if (message instanceof  TextMessage){
                    try {
                        String text = ((TextMessage) message).getText();
                        System.out.println(text+"接收的消息！");
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        consumer.close();
        session.close();
    }
}

producer提交时的事务

事务开启 true

只执行send并不会提交到队列中，只有当执行session.commit()时，消息才被真正的提交到队列中。

事务不开启 false

只要执行send，就进入到队列中。

consumer 接收时的事务

事务开启，签收必须写

Session.SESSION_TRANSACTED

收到消息后，消息并没有真正的被消费。消息只是被锁住。一旦出现该线程死掉、抛异常，或者程序执行了session.rollback()那么消息会释放，重新回到队列中被别的消费端再次消费。

事务不开启，签收方式选择

Session.AUTO_ACKNOWLEDGE

只要调用comsumer.receive方法 ，自动确认。

事务不开启，签收方式选择

Session.CLIENT_ACKNOWLEDGE

需要客户端执行 message.acknowledge(),否则视为未提交状态，线程结束后，其他线程还可以接收到。

  这种方式跟事务模式很像，区别是不能手动回滚,而且可以单独确认某个消息。

手动签收

事务不开启，签收方式选择

Session.DUPS_OK_ACKNOWLEDGE

在Topic模式下做批量签收时用的，可以提高性能。但是某些情况消息可能会被重复提交，使用这种模式的consumer要可以处理重复提交的问题。