码迷,mamicode.com
首页 > 其他好文 > 详细

ActiveMQ的使用

时间:2020-03-01 19:31:09      阅读:69      评论:0      收藏:0      [点我收藏+]

标签:pid   cte   tcp   connect   enc   ups   apach   package   throw   

实现点对点通讯模式

  1.导入依赖

  <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-core</artifactId>
      <version>5.7.0</version>
    </dependency>

  2.创建生产者

技术图片
package com.wn.ddd;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import javax.jms.*;

public class Producter {
    public static void main(String[] args) throws JMSException {
        // ConnectionFactory :连接工厂,JMS 用它创建连接
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
        // JMS 客户端到JMS Provider 的连接
        Connection connection = connectionFactory.createConnection();
        //启动连接
        connection.start();
        // Session: 一个发送或接收消息的线程    false:代表不带事务的session   AUTO_ACKNOWLEDGE:代表自动签收
        Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        // Destination :消息的目的地;消息发送给谁.
        // 获取session注意参数值my-queue是Query的名字
        Queue queue = session.createQueue("my-queue");
        // MessageProducer:创建消息生产者
        MessageProducer producer = session.createProducer(queue);
        // 设置不持久化  PERSISTENT:代表持久化  NON_PERSISTENT:代表不持久化
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        // 发送消息
        for (int i = 1; i <= 5; i++) {
            sendMsg(session, producer, i);
        }
        System.out.println("发送成功!");
        session.close();
        connection.close();
    }
    /**
     * 在指定的会话上,通过指定的消息生产者发出一条消息
     *
     * @param session
     *            消息会话
     * @param producer
     *            消息生产者
     */
    public static void sendMsg(Session session, MessageProducer producer, int i) throws JMSException {
        // 创建一条文本消息
        TextMessage message = session.createTextMessage("Hello ActiveMQ!" + i);
        // 通过消息生产者发出消息
        producer.send(message);
    }
}
技术图片

  3.创建消费者

技术图片
package com.wn.ddd;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class JmsReceiver {
    public static void main(String[] args) throws JMSException {
        // ConnectionFactory :连接工厂,JMS 用它创建连接
        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,
                ActiveMQConnection.DEFAULT_PASSWORD, "tcp://127.0.0.1:61616");
        // JMS 客户端到JMS Provider 的连接
        Connection connection = connectionFactory.createConnection();
        connection.start();
        // Session: 一个发送或接收消息的线程  true:表单开启事务  AUTO_ACKNOWLEDGE:代表自动签收
        Session session = connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE);
        // Destination :消息的目的地;消息发送给谁.
        // 获取session注意参数值xingbo.xu-queue是一个服务器的queue,须在在ActiveMq的console配置
        Queue queue = session.createQueue("my-queue");
        // 消费者,消息接收者
        MessageConsumer consumer = session.createConsumer(queue);
        while (true) {
            //receive():获取消息
            TextMessage message = (TextMessage) consumer.receive();
            if (null != message) {
                System.out.println("收到消息:" + message.getText());
            } else {
                break;
            }
        }
        //回收资源
        session.close();
        connection.close();
    }
}
技术图片

  4.测试

    4.1 首先需启动生产者

      技术图片

      控制台执行完成后,去查看一下队列的情况

      技术图片

     4.2 启动消费者

      技术图片

      可以再次查看队列的情况

      技术图片

JMS消息可靠机制

  ActiveMQ消息签收机制:

    客户端成功接收一条消息的标志是一条消息被签收,成功应答。

  消息的签收情况分为两种:

    1.带事务的session

      如果session带有事务,并且事务成功提交,则消息被自动签收。如果事务回滚,则消息会被再次传递。

    2.不带事务的session

      不带事务的session签收方式,取决于session的配置。

  ActiveMQ支持以下三种模式:

    1.Session.AUTO_ACKNOWLEDGE  消息自动签收

    2.Session.CLIENT_ACKNOWLEDGE  客户端调用acknowledge方法手动签收

     textMessage.acknowledge()  手动签收

    3.Session.DUPS_OK_ACKNOWLEDGE  不是必须签收,消息可能会重复发送。在第二次重新传送消息的时候,消息只有在被确认之后,才认为已经被成功地消费了。

发布/订阅(Pub/Sub)

  1.消费者

技术图片
package com.wn.fndy;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

//消费者
public class TopReceiver {
    private static String BROKERURL = "tcp://127.0.0.1:61616";
    private static String TOPIC = "my-topic";

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

    private static void start() throws JMSException {
        System.out.println("消费者启动......");
        // 创建ActiveMQConnectionFactory 会话工厂
        ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory(
                ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, BROKERURL);
        Connection connection = activeMQConnectionFactory.createConnection();
        // 启动JMS 连接
        connection.start();
        // 不开消息启事物,消息主要发送消费者,则表示消息已经签收
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        // 创建一个队列
        Topic topic = session.createTopic(TOPIC);
        MessageConsumer consumer = session.createConsumer(topic);
        // consumer.setMessageListener(new MsgListener());
        while (true) {
            TextMessage textMessage = (TextMessage) consumer.receive();
            if (textMessage != null) {
                System.out.println("接受到消息:" + textMessage.getText());
                // textMessage.acknowledge();// 手动签收
                // session.commit();
            } else {
                break;
            }
        }
        connection.close();
    }

}
技术图片

  2.生产者

技术图片
package com.wn.fndy;

import com.sun.scenario.effect.impl.sw.sse.SSEBlend_SRC_OUTPeer;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

//生产方
public class TOPSend {
    // tcp 地址
    public static final String BROKERURL = "tcp://localhost:61616";
    // 目标,在ActiveMQ管理员控制台创建 http://localhost:8161/admin/queues.jsp
    public static final String TOPIC = "my-topic";

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

    private static void start() throws JMSException {
        System.out.println("生产者已经启动......");
        //创建ActiveMQConnectionFactory会话工厂
        ActiveMQConnectionFactory activeMQConnectionFactory=new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, BROKERURL);
        Connection connection = activeMQConnectionFactory.createConnection();
        //启动JMS连接
        connection.start();
        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
        MessageProducer producer = session.createProducer(null);
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        send(producer,session);
        System.out.println("发送成功!");
        connection.close();
    }

    private static void send(MessageProducer producer, Session session) throws JMSException {
        for (int i=1;i<=5;i++){
            System.out.println("我是消息"+i);
            TextMessage textMessage = session.createTextMessage("我是消息" + i);
            Destination destination = session.createTopic(TOPIC);
            producer.send(destination,textMessage);
        }
    }


}
技术图片

  3.测试

    3.1 启动两个消费者,因为只有消费者先订阅,生产者才会发送消息,消费者才能接收消息;(发布/订阅是一个发送者,对应多个消费者)

      技术图片

       技术图片

    3.2 启动生产者

      技术图片

      技术图片

       查看一个控制台中两个消费者的情况

      技术图片技术图片

ActiveMQ的使用

标签:pid   cte   tcp   connect   enc   ups   apach   package   throw   

原文地址:https://www.cnblogs.com/mayuan01/p/12391357.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!