以项目为牵引   全程实训   以就业为导向  以动手能力为目标

一样的教育    不一样的结果     全国咨询热线:400-078-8286

ActiveMQ_3Java实现

2018-12-22 来源:桂林志成IT教育中心  点击量:1798
摘要:ActiveMQ_3Java实现

添加相应的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 = newActiveMQConnectionFactory("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 = newActiveMQConnectionFactory("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对象(topic、queue),此处创建一个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);

热门标签:Java 实现

热门课程

相关文章>

合作伙伴