zoukankan      html  css  js  c++  java
  • activeMQ的回顾

    JMS:

    JMS基本概念:

    JMS(Java Message Service) 即Java消息服务。它提供标准的产生、发送、接收消息的接口简化企业应用的开发。它支持两种消息通信模型:点到点(point-to-point)(P2P)模型和发布/订阅(Pub/Sub)模型。

    1)点对点方式(point-to-point)

    点对点的消息发送方式主要建立在 Message Queue,Sender,Receiver上,Message Queue 存贮消息,Sender发送消息,Receiver接收消息.具体点就是Sender Client发送Message 到Queue中 ,而Receiver Client从Queue中接收消息和发送消息已接受到Quere,确认消息接收。消息发送客户端与接收客户端没有时间上的依赖,发送客户端可以在任何时刻发送信息到Queue,而不需要知道接收客户端是不是在运行

    2)发布/订阅 方式(publish / subscribe)

    发布/订阅方式用于多接收客户端的方式.作为发布订阅的方式,可能存在多个接收客户端,并且接收端客户端与发送客户端存在时间上的依赖。一个接收端只能接收他创建以后发送客户端发送的信息。作为subscriber ,在接收消息时有两种方法,destination的receive方法,和实现message listener 接口的onMessage 方法。

    几个重要概念:

    Destination:消息发送的目的地,也就是所谓的Queue和Topic。创建好一个消息之后,只需要把这个消息发送到目的地,消息的发送者就可以继续做自己的事情,而不用等待消息被处理完成。至于这个消息什么时候,会被哪个消费者消费,完全取决于消息的接受者。 Message:从字面上就可以看出是被发送的消息。它有下面几种类型:

    StreamMessage:Java 数据流消息,用标准流操作来顺序的填充和读取。

    MapMessage:一个Map类型的消息;名称为 string 类型,而值为 Java 的基本类型。 TextMessage:普通字符串消息,包含一个String。

    ObjectMessage:对象消息,包含一个可序列化的Java 对象 BytesMessage:二进制数组消息,包含一个byte[]。 XMLMessage: 一个XML类型的消息。

    最常用的是TextMessage和ObjectMessage。

    Session:与JMS提供者所建立的会话,通过Session我们才可以创建一个Message。

    Connection:与JMS提供者建立的一个连接。可以从这个连接创建一个会话,即Session。 ConnectionFactory:那如何创建一个Connection呢?这就需要下面讲到的

    ConnectionFactory了。通过这个工厂类就可以得到一个与JMS提供者的连接,即Conection。

    Producer:消息的生产者,要发送一个消息,必须通过这个生产者来发送。

    MessageConsumer:与生产者相对应,这是消息的消费者或接收者,通过它来接收一个消息。

    ActiveMQ:

    ActiveMQ 是Apache出品,能力强劲的开源的消息总线。ActiveMQ 是一个完全支持JMS和J2EE规范的 JMS Provider实现。

    在官方地址: http://activemq.apache.org/上可以下载到ActiveMQ,我们使用的是ActiveMQ5.2版本,下载apache-activemq-5.2.0-bin.zip到本地,解压缩之后即可使用。在相对应解压目录下的apache-activemq-5.2.0\bin\中找到activemq.bat运行它,ActiveMQ就运行起来了。需要注意的是,在运行ActiveMQ之前需要在环境变量中配置好

    JAVA_HOME ,PATH和CLASSPATH。

    ActiveMQ的默认的监听端口为61616,可以在activemq.xml文件中进行修改,建议不要进行修改。另外在ActiveMQ启动之后,可以在ActiveMQ的控制台http://localhost:8161/admin/中查看运行的状态,进行各种操作。

    一个典型的JMS程序的创建过程:

    1.通过jndi查询ConnectionFactory(这个jndi的配置是针对具体的jms提供者的,如ActiveMQ 就使用org.apache.activemq.ActiveMQConnectionFactory) 2.使用ConnectionFactory创建一个Connnection。 3.启动Connection。

    4.使用Connection创建一个或者多个Session 5.通过jndi查询一个或者多个Destination。

    6.使用session和Destination创建对应的MessageProducer和MessageConsumer。 按照上面过程得到这样一个简图,

    ConnectionFactory---->Connection--->Session--->Message

    Destination + Session------------------------------------>Producer

    Destination + Session------------------------------------>MessageConsumer

    我们使用ActiveMQ去实现这个简单的JMS程序: ? 首先需要得到ConnectionFactoy。 ConnectionFactory factory = new ActiveMQConnectionFactory(url); ? 然后又ConnectionFactory创建一个Connection, 再启动这个Connection: Connection connection = factory.createConnection(); connection.start(); ? 接下来需要由Connection创建一个Session: Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE) ? 获得Destination,这里创建一个Queue作为Destination。 Destination destination = session.createQueue(? 下面就可以创建Message了,这里创建一个TextMessage。 Message message = session.createTextMessage(? 要想把刚才创建的消息发送出去,需要由Session和Destination创建一个消息生产者: MessageProducer producer = session.createProducer(queue); ? 下面就可以发送刚才创建的消息了: producer.send(message); ? 消息发送完成之后,我们需要创建一个消息消费者来接收这个消息: MessageConsumer comsumer = session.createConsumer(queue); Message recvMessage = comsumer.receive(); ? 消息消费者接收到这个消息之后,就可以得到它的内容: System.out.println(((TextMessage)recvMessage).getText());

    从而得到这个简单的JMS的程序如下,在运行这个程序之前,你需要在所建工程中导入2个jar包javax.jms.jar(可以在网上下载到)和activemq-all-5.2.0.jar(在/apache-activemq-5.2.0/下) import javax.jms.*; import org.apache.activemq.ActiveMQConnectionFactory;

    public class MessageSendAndReceive {

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

    ConnectionFactory factory = new ActiveMQConnectionFactory(

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

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

    Destination destination = session.createQueue( Message message = session.createTextMessage( MessageProducer producer = session .createProducer(destination); producer.send(message);

    System.out.println( MessageConsumer comsumer = session .createConsumer(destination);

    Message recvMessage = comsumer.receive(); System.out.println(

    + ((TextMessage) recvMessage).getText()); } }

    启动ActiveMQ后,运行上面的程序将会得到如下结果:

    Send Message Completed! Received:Hello JMS!

    小实验:在前面介绍到,接收消息时有两种方法,destination的receive方法,和实现message listener 接口的onMessage 方法。现在我们要编写一个消息监听者,监听队列,当队列里一旦有消息则自动接收队列里的消息。 import javax.jms.*; import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnectionFactory; public class Jms_Receiver_Listener extends Thread implements MessageListener, ExceptionListener { // ConnectionFactory :连接工厂,JMS 用它创建连接 ConnectionFactory connectionFactory; // Connection :JMS 客户端到JMS Provider 的连接 Connection connection = null; // Session: 一个发送或接收消息的线程 Session session;

    // Destination :消息的目的地 Destination destination; //消息接收者

    MessageConsumer consumer;

    public Jms_Receiver_Listener() {

    String url =

    // 在ActiveMq的console配置的queue的名字 String queue =

    connectionFactory = new ActiveMQConnectionFactory( ActiveMQConnection.DEFAULT_USER,

    ActiveMQConnection.DEFAULT_PASSWORD, url); // 构造从工厂得到连接对象 try {

    connection = connectionFactory.createConnection(); connection.setExceptionListener(this);// 异常处理 connection.start();// 连接启动

    //如果为true,则队列里面的消息没有被取走,继续存在 session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // 获取session,

    destination = session.createQueue(queue);

    consumer = session.createConsumer(destination); } catch (JMSException e) {

    System.err.println( e.printStackTrace(); } }

    public void run() { try {

    consumer.setMessageListener(this); } catch (JMSException e) { System.err

    .println( e.printStackTrace(); } }

    public void onMessage(Message message) { try {

    if (message instanceof TextMessage) {

    TextMessage txtMsg = (TextMessage) message; String msg = txtMsg.getText();

    }

    System.out.println( }

    } catch (JMSException e) { System.err

    .println( e.printStackTrace(); } }

    // 异步消息异常处理

    public void onException(JMSException arg0) { System.err.println(异常!}

    //测试程序

    public static void main(String[] args) {

    Jms_Receiver_Listener jrl = new Jms_Receiver_Listener(); jrl.start(); }

    运行上面的程序,一旦队列TestQueue队列中有消息,监听就会从队列中取出消息,打印到控制台上。

    Received: 学习ActiveMQ 发送消息 1 Received: 学习ActiveMQ 发送消息2 Selector:

    前面用到的消息接收者的创建方式是:session.createConsumer(destination),这种方式,消息的接收者会把所有消息接收到,而如果只要选择性的接受你想要的消息,就需要使用consumer = session.createConsumer(destination,selector)的方式。

    假设,有多个消息的监听者同时监听同一个队列,使用这种方式就可以根据selector来对队列中的消息进行过滤,只有符合selector的消息才会从队列中被取出来,不符合的消息还保留在队列中。

    首先在消息的发送方,发送消息之前对消息设置属性:

    message.setStringProperty(这样在消息接收方使用selector的方式为:

    consumer = session.createConsumer(destination,’ 其他地方不需要做任何变化,这样消息的监听者只会接收MessageID为1001的消息,而其他消息不会取出,仍然放在队列中。

  • 相关阅读:
    用shell脚本监控进程是否存在 不存在则启动的实例
    vue+element-ui+ajax实现一个表格的实例
    c的指针和php中的引用的区别
    mysql or条件查询优化
    Linq查询操作之投影操作
    Linq查询操作之Where筛选
    Linq专题之查询操作
    linq之join子句
    linq之let子句
    linq之into子句
  • 原文地址:https://www.cnblogs.com/bingru/p/7565426.html
Copyright © 2011-2022 走看看