ActiveMQ_3Java实现

2018-12-13 08:55:59来源:博客园 阅读 ()

新老客户大回馈,云服务器低至5折

Java实现

 

添加相应的jar包

<dependency>

  <groupId>org.apache.activemq</groupId>

  <artifactId>activemq-all</artifactId>

  <version>x.xx.x</version>

</dependency>

 

创建生产者类(点对点)

public class ProducerTest {

    // 异步发送asyn

    // 死信队列 DLQ

    // 文件上传

    // header properties使用

    // jdbc存储

    // byteMsg objMsg inputMsg

    // mq ptp 的使用场景

    // mq中所有的队列名以及每个队列中未被消费的消息数量

    // mq sub/pub 的使用场景

    @Test

    public void testQueueProducer() throws JMSException{

       // 第一步:创建ConnectionFactory对象,需要指定服务端ip及端口号。

       //brokerURL服务器的ip及端口号

       ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.114.129:61616");

       // 第二步:使用ConnectionFactory对象创建一个Connection对象

       Connection connection = connectionFactory.createConnection();

       // 第三步:开启连接,调用Connection对象的start方法

       connection.start();

       // 第四步:使用Connection对象创建一个Session对象。

       //第一个参数:是否开启事务。true:开启事务,第二个参数忽略。

       //第二个参数:当第一个参数为false时,才有意义。消息的应答模式。1、自动应答2、手动应答。一般是自动应答

       Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

       // 第五步:使用Session对象创建一个Destination对象(topic、queue),此处创建一个Queue对象。

       //参数:队列的名称。

       Queue queue = session.createQueue("test-queue");

       // 第六步:使用Session对象创建一个Producer对象。

       MessageProducer producer = session.createProducer(queue);

       producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

       // 第七步:创建一个Message对象,创建一个TextMessage对象。

       /*TextMessage message = new ActiveMQTextMessage();

       message.setText("hello activeMq,this is my first test.");*/

       TextMessage textMessage = session.createTextMessage("hello activeMq,this is my first test.");

       producer.send(textMessage);

       producer.close();

       session.close();

       connection.close();

    }

}

 

创建消费者类(点对点)

public class ConsumerTest {

    @Test

    public void testQueueConsumer() throws JMSException, IOException{

       // 第一步:创建一个ConnectionFactory对象。

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://192.168.114.129:61616");

        // 第二步:从ConnectionFactory对象中获得一个Connection对象。

        Connection connection = connectionFactory.createConnection();

        // 第三步:开启连接。调用Connection对象的start方法。

        connection.start();

        // 第四步:使用Connection对象创建一个Session对象。

        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        // 第五步:使用Session对象创建一个Destination对象。和发送端保持一致queue,并且队列的名称一致。

        Queue queue = session.createQueue("queue.stu");

        // 第六步:使用Session对象创建一个Consumer对象。

        MessageConsumer consumer = session.createConsumer(queue);

        // 第七步:接收消息。

        consumer.setMessageListener(new MessageListener()

             @Override

             public void onMessage(Message message) {

                  try {

                    //if (TextMessage.class.isAssignableFrom(message.getClass()))

                    if (message instanceof TextMessage) {

                          TextMessage textMessage = (TextMessage) message;

                          String text = null;

                          //取消息的内容

                          text = textMessage.getText();

                          // 第八步:打印消息。

                          System.out.println(text);

                    }

                  } catch (JMSException e) {

                      e.printStackTrace();

                  }

             }

        });

        //等待键盘输入

        System.in.read();

        // 第九步:关闭资源

        consumer.close();

        session.close();

        connection.close();

    }

}

 

运行active服务器验证实现情况

 

发布(对点对点代码修改 创建Topic Destination)

// 第五步:使用Session对象创建一个Destination对象(topicqueue),此处创建一个topic对象。

// 参数:话题的名称。

Topic topic = session.createTopic("test-topic");

// 第六步:使用Session对象创建一个Producer对象。

MessageProducer producer = session.createProducer(topic);

 

订阅(对点对点代码修改 创建Topic Destination)

// 第五步:使用Session对象创建一个Destination对象。和发送端保持一致topic,并且话题的名称一致。

Topic topic = session.createTopic("test-topic");

// 第六步:使用Session对象创建一个Consumer对象。

MessageConsumer consumer = session.createConsumer(topic);

 

标签:

版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有

上一篇:杂谈篇之我是怎么读源码的,授之以渔

下一篇:如何使用 eclipse进行断点 debug 程序