ActiveMQ学习-Api介绍 (4)

Session

第一节提到了消息提供者在创建session的时候第一个参数是事务的意思,如果将其改成 true ,那么当发送完消息后就要 commit() 一下,消息才能发出

public class Producer {

  public static void main(String[] args) throws JMSException {
    // 省略了的代码...

    // 如果session不支持事件,就是FALSE,支持事务就是true
    Session session = connection.createSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

    // 省略了的代码...

    // 上面开启了事务,这里就要提交一下
    session.commit();

    // 当然如果上面发送消息出问题了,也可以通过 rollback() 回滚掉
    // session.rollback();
  }
}

消息消费者在消费消息的时候 session 的事务就没有意义了,true, false 都一样的消费,不过为了保持一致,还是建议都写成一样的,既:提供者开启事务,消费者也开启事务

创建session的时候,第二个参数是签收模式,有以下几种

TextMessage.acknowledge()

MessageProducer

MessageProducer的send方法最多有5个参数

void send(
  Destination destination,
  Message message,
  int deliveryMode,
  int priority,
  long timeToLive)
  throws JMSException;
  • destination 目的地,消息要发送到哪去,通过session创建 Destination destination = session.createQueue("amq-demo");
  • message 消息内容,它是一个接口,实现有以下几种消息类型 ActiveMQ学习-Api介绍 (4)
  • deliveryMode 消息的传输模式
    • DeliveryMode.NON_PERSISTENT 消息不做持久化
    • DeliveryMode.PERSISTENT 消息持久化
  • priority 消息优先级,有0-9十个级别,0-4为普通消息,5-9是加急消息,默认为4
    • 优先级开启要在 activemq.xml 里进行配置才能生效,找到 policyEntries 标签,在里面加上 <policyEntry queue="amq-demo" prioritizedMessages="true" useCache="false" expireMessagesPeriod="0" queuePrefetch="1"/> 即可
  • timeToLive 消息存活时间,默认一直存活

MessageConsumer

MessageConsumer是session创建的,一般给一个Destination即可,不过它还有两个参数

// 创建队列消费者
MessageConsumer createConsumer(
  Destination destination,
  java.lang.String messageSelector,
  boolean NoLocal)
  throws JMSException;

// 创建主题消费者
TopicSubscriber createDurableSubscriber(
  Topic topic,
  String name,
  String messageSelector,
  boolean noLocal)
  throws JMSException;
MapMessages

这里说一下MessageSelector

举见个例子

// 选择名字为 z3 的消息
String selector1 = "name = 'z3'";
// 选择年龄大于20的消息
String selector2 = "age > 20";
// 选择名字为 z3 和 年龄大于20的消息
String selector3 = "name = 'z3' AND age > 20";

MessageConsumer consumer = session.createConsumer(destination, selector1);

消息的接收方式,通过 receive() 方法来接收,该方法是阻塞的,一般会用一个监听来实现

HelloWorld里接收方式是通过 while(true){} 来实现的,这种代码不推荐

while(true) {
  TextMessage msg = (TextMessage) consumer.receive();
  System.out.println("消费数据:" + msg.getText());
}

下面是通过监听的方式来接收消息

public class Consumer {

  public static void main(String[] args) throws JMSException {
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
        ActiveMQConnectionFactory.DEFAULT_USER,
        ActiveMQConnectionFactory.DEFAULT_PASSWORD,
        "tcp://localhost:61616"
    );

    Connection connection = connectionFactory.createConnection();
    connection.start();

    Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

    Destination destination = session.createQueue("amq-demo");
    MessageConsumer consumer = session.createConsumer(destination);

    consumer.setMessageListener(new MsgListener());

//    while(true) {
//      TextMessage msg = (TextMessage) consumer.receive();
//      System.out.println("消费数据:" + msg.getText());
//    }
  }

  static class MsgListener implements MessageListener {

    public void onMessage(Message message) {
      try {
        System.out.println("消费数据:" + ((TextMessage) message).getText());
      } catch (JMSException e) {
        e.printStackTrace();
      }
    }
  }
}

看起来是不是优雅多了 : )

创建临时消息

消息提供者还可以创建临时消息,通过 session.createTemporaryQueue() 方法创建,这种方式创建的消息,当connection关闭之后,消息也就没有了

发布订阅

除了队列,ActiveMQ还可以创建主题,消费者可以通过订阅主题来消费消息,用法跟队列基本一致,下面上代码

提供者

public class Producer {

  public static void main(String[] args) throws JMSException {
    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
        ActiveMQConnectionFactory.DEFAULT_USER,
        ActiveMQConnectionFactory.DEFAULT_PASSWORD,
        "tcp://localhost:61616"
    );

    Connection connection = connectionFactory.createConnection();
    connection.start();

    // 如果session不支持事件,就是FALSE,支持事务就是true
    Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
    Destination destination = session.createTopic("amq-topic-demo");

    MessageProducer producer = session.createProducer(null);

    for (int i = 0; i < 100; i++) {
      TextMessage msg = session.createTextMessage("生产消息" + i);
      producer.send(destination, msg);
    }

//    session.commit();
//    session.rollback();

    connection.close();
  }
}

消费者1 与 消费者2 是一样的,把日志内容改一下就可以了

public class Consumer1 {

  public static void main(String[] args) throws JMSException {

    ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
        ActiveMQConnectionFactory.DEFAULT_USER,
        ActiveMQConnectionFactory.DEFAULT_PASSWORD,
        "tcp://localhost:61616"
    );

    Connection connection = connectionFactory.createConnection();
    connection.start();

    Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);

    Destination destination = session.createTopic("amq-topic-demo");
    MessageConsumer consumer = session.createConsumer(destination);

    consumer.setMessageListener(new MsgListener());

  }

  static class MsgListener implements MessageListener {

    public void onMessage(Message message) {
      try {
        System.out.println("消费者1 消费数据:" + ((TextMessage) message).getText());
      } catch (JMSException e) {
        e.printStackTrace();
      }
    }
  }
}

这个启动就要先启动所有的消费者,然后再启动提供者,这样提供者发布的主题消息,消费者才能进行消费,运行一下,可以看到消费者1跟消费者2都消费了提供者提供的消息

原文链接:

原文 

https://tomoya92.github.io/2018/11/08/activemq-api/

本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。

PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » ActiveMQ学习-Api介绍 (4)

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址