自己动手实现消息队列之JMS

什么是JMS?JMS的诞生史?

在JMS还没有诞生前，每个企业都会有自己的一套内部消息系统，比如项目组A需要调用到项目组B的系统，项目组B也有可能会调用到项目组C的系统。这样每个公司都有自己的一套实现。很不规范，所以Apache基金会，为企业消息产品专门定义了一套规范。我们可以把JMS当作是一系列接口及相关语义的集合，通过这些接口和语义定义了JSM客户端如何去访问消息系统。简单点来说就是JMS主要干了两件事，定义通用的消息格式，和消息传递的模式。

体系结构

JMS由以下元素组成。[1]

JMS提供者

连接面向消息中间件的，JMS接口的一个实现。提供者可以是Java平台的JMS实现，也可以是非Java平台的面向消息中间件的适配器。

JMS客户

生产或消费基于消息的Java的应用程序或对象。

JMS生产者

创建并发送消息的JMS客户。

JMS消费者

接收消息的JMS客户。

JMS消息

包括可以在JMS客户之间传递的数据的对象

JMS队列

一个容纳那些被发送的等待阅读的消息的区域。与队列名字所暗示的意思不同，消息的接受顺序并不一定要与消息的发送顺序相同。一旦一个消息被阅读，该消息将被从队列中移走。

JMS主题

一种支持发送消息给多个订阅者的机制。

对象模型

JMS对象模型包含如下几个要素：[2]

1）连接工厂。连接工厂（ConnectionFactory）是由管理员创建，并绑定到JNDI树中。客户端使用JNDI查找连接工厂，然后利用连接工厂创建一个JMS连接。

2）JMS连接。JMS连接（Connection）表示JMS客户端和服务器端之间的一个活动的连接，是由客户端通过调用连接工厂的方法建立的。

3）JMS会话。JMS会话（Session）表示JMS客户与JMS服务器之间的会话状态。JMS会话建立在JMS连接上，表示客户与服务器之间的一个会话线程。

4）JMS目的。JMS目的（Destination），又称为消息队列，是实际的消息源。

5）JMS生产者和消费者。生产者（Message Producer）和消费者（Message Consumer）对象由Session对象创建，用于发送和接收消息。

6）JMS消息通常有两种类型：

① 点对点（Point-to-Point）。在点对点的消息系统中，消息分发给一个单独的使用者。点对点消息往往与队列（javax.jms.Queue）相关联。

② 发布/订阅（Publish/Subscribe）。发布/订阅消息系统支持一个事件驱动模型，消息生产者和消费者都参与消息的传递。生产者发布事件，而使用者订阅感兴趣的事件，并使用事件。该类型消息一般与特定的主题（javax.jms.Topic）关联。

上面都是一些概念型的东西，有个大概就行了，真正想理解还得一边写代码一遍思考；

下面将简单模拟一下一个基于JMS规范的消息队列：

定义JMS消息

JMS 定义了5中消息类型： TextMessage、MapMessage、BytesMessage、

StreamMessage和ObjectMessage。

  • TextMessage（文本消息）

将数据作为简单字符串存放在主体中(XML就可以作为字符串发)

  • MapMessage（映射消息）

使用一张映射表来存放其主体内容（参照Jms API）

  • BytesMessage（字节消息）

将字节流存放在消息主体中。适合于下列情况：必须压缩发送的大量数据、需要与现有

消息格式保持一致等（参照Jms API）

  • StreamMessage（流消息）

用于处理原语类型。这里也支持属性字段和MapMessage所支持的数据类型。使用这种

消息格式时，收发双方事先协商好字段的顺序，以保证写读顺序相同（参照Jms API）

  • ObjectMessage（对象消息）

用于往消息中写入可序列化的对象。

消息中可以存放一个对象，如果要存放多个对象，需要建立一个对象集合，然后把这个

集合写入消息。

这里简单定义一下TextMessage（文本消息）。在我们的程序中，发送和接收消息的都只是一个字符串而已，所以定义很简单。String就OK。

JMS队列：

这里使用我们Java的 LinkedList队列集合去实现。

// 消息队列
private static LinkedList<String> jmsQueue=new LinkedList<String>(); 

JMS客户，生产者，消费者

在JMS里面的客户，并不是我们的消费者，这里指的是基于消息的Java的应用程序或者对象。这句话什么意思啦，按照我的理解来说，就是我有一个程序用到了基于消费者产生的数据，那么我就是消费者的客户，因为我在使用消费者。这个很类似于我们的客户端，比较简单。

下面主要讲讲生产者和消费者：

这里主要有两种消息订阅模型，一种是点对点，即一个消费者和一个生产者之间进行传输消息，一个是多对多，即会有多个消费者和生产者之间进行消息传输。

Point-to-Point(P2P)
Publish/Subscribe(Pub/Sub)

  1. P2P

    1. P2P模式图 
    2. 涉及到的概念 
      1. 消息队列（Queue）
      2. 发送者(Sender)
      3. 接收者(Receiver)
      4. 每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。
    3. P2P的特点
      1. 每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中)
      2. 发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列
      3. 接收者在成功接收消息之后需向队列应答成功

      如果你希望发送的每个消息都应该被成功处理的话，那么你需要P2P模式。

  1. Pub/Sub

    1. Pub/Sub模式图 
    2. 涉及到的概念 
      1. 主题（Topic）
      2. 发布者（Publisher）
      3. 订阅者（Subscriber） 
        客户端将消息发送到主题。多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。
    3. Pub/Sub的特点
      1. 每个消息可以有多个消费者
      2. 发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息，而且为了消费消息，订阅者必须保持运行的状态。
      3. 为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。

      如果你希望发送的消息可以不被做任何处理、或者被一个消息者处理、或者可以被多个消费者处理的话，那么可以采用Pub/Sub模型

下面创建一下这几个模型：

首先创建一个消费者生产者模式的缓冲区来做我们的多线程消息队列管理，用来接收数据和传输数据；

/**
 * 实现缓冲区
 *
 * @author gh
 *
 */
public class JmsBuffer {
	// 队列 最大存储量
	private final static int  MAX_SIZE = 100;
	// 消息队列
	private static LinkedList<String> jmsQueue=new LinkedList<String>();
	static JmsBuffer buffer;
	 // 生产消息
    public static void produce(String str)
    {
        // 同步代码段
        synchronized (jmsQueue)
        {
            // 如果仓库剩余容量不足
            while (jmsQueue.size()> MAX_SIZE)
            {
                System.out.println("你要生产的消息为" + "/t【库存量】:"
                        + jmsQueue.size() + "/t暂时不能执行生产任务!");
                try
                {
                    // 由于条件不满足，生产阻塞
                	jmsQueue.wait();
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }  

            // 生产条件满足情况下，生产消息  

                jmsQueue.add(str);  

           System.out.println("已经生产该消息" + str + "/t【现仓储量为】:" + jmsQueue.size());  

            jmsQueue.notifyAll();
        }
    }  

    // 消费消息
    public static String consume()
    {
        // 同步代码段
        synchronized (jmsQueue)
        {
            // 如果仓库存储量不足
            while (jmsQueue.size() > MAX_SIZE)
            {
                System.out.println("【消息库存量】:"
                        + jmsQueue.size() + "/t暂时不能执行生产任务!");
                try
                {
                    // 由于条件不满足，消费阻塞
                	jmsQueue.wait();
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }  

            // 消费条件满足情况下，消费该消息 

                String str=(String) jmsQueue.removeLast();   

            System.out.println("【已经消费该消息】:" + str+ "/t【现仓储量为】:" + jmsQueue.size());  

           jmsQueue.notifyAll();
           return str;
        }
    }
    public synchronized static JmsBuffer getJmsBuffer(){
    	if(buffer==null){
    		return new JmsBuffer();
    	}else{
    		return buffer;
    	}
    }

}

这是一个很经典的消费者生产者模型，

wait() / nofity()方法是基类Object的两个方法，也就意味着所有Java类都会拥有这两个方法，这样，我们就可以为任何对象实现同步机制。

wait()方法：当缓冲区已满/空时，生产者/消费者线程停止自己的执行，放弃锁，使自己处于等等状态，让其他线程执行。

notify()方法：当生产者/消费者向缓冲区放入/取出一个产品时，向其他等待的线程发出可执行的通知，同时放弃锁，使自己处于等待状态。

服务端

服务端用来接收数据和传输数据：

public class Server extends Thread {
	private final Socket client;
	public Server(Socket c) {
		this.client = c;
	}

	public Socket getClient() {
		return client;
	}

	@Override
	public void run() {
		try {
			BufferedReader in = new BufferedReader(new InputStreamReader(
					client.getInputStream()));
			PrintWriter out = new PrintWriter(client.getOutputStream());
			while (true) {
				String str = in.readLine();
				if(str.contains("code:200")){ //200代表生产消息
					//生产消息
					JmsBuffer.produce(str);
				}
				if(str.contains("code:400")){ //400代表消费消息
					String messag=JmsBuffer.consume();
			             out.println(messag);
			             out.flush();
				}
				System.out.println(str);
				out.flush();
				if (str.equals("end"))
					break;
			}
			client.close();
		} catch (IOException ex) {
		} finally {
		}
	}
	public static void main(String[] args) throws IOException {
		ServerSocket server = new ServerSocket(5678);
		while (true) {
			// transfer location change Single User or Multi User

			Server mc = new Server(server.accept());
			mc.start();
		}
	}
}

服务器基本步骤：

1.指定端口实例化一个SeverSocket

2.调用ServerSocket的accept()方法，以在等待连接期间造成阻塞

3.获取位于该底层的Socket的流以进行读写操作

4.将数据封装成流

5.对Socket进行读写

6.关闭打开的流

服务器端会根据CODE码来确认是否需要读取数据，在消费者生产者模型里面也叫消费数据。

Client端

public class Client {
	static Socket server;
	public static void creaSockeet() throws UnknownHostException, IOException{
		server = new Socket(InetAddress.getLocalHost(), 5678);
	}
	//发布消息
    public static void pub(String message) throws UnknownHostException, IOException{
         BufferedReader in = new BufferedReader(new InputStreamReader(
                 server.getInputStream()));
         PrintWriter out = new PrintWriter(server.getOutputStream());
             out.println(message);
             out.flush();
             System.out.println(in.readLine());
         server.close();
    }
  //发布消息
    public static String sub(String code) throws UnknownHostException, IOException{
         BufferedReader in = new BufferedReader(new InputStreamReader(
                 server.getInputStream()));
         PrintWriter out = new PrintWriter(server.getOutputStream());
             out.println(code);
             out.flush();
             String str = in.readLine();
             System.out.println(str);
             System.out.println("获取的消息为:"+str);
         //server.close();
         return str;
    }

}

客户端基本步骤：

1.通过IP地址和端口实例化Socket，请求连接服务器

2.获得Socket上的流以进行读写

3.把流封装进BufferedReader/PrintWriter的实例

4.对Socket进行读写

5.关闭打开的流

客户端里面有两个方法，叫做发布和订阅，符合我们的生产者消费者原形。

最后创建一个生产者，一个消费者去测试。先启动我们的Server端。

public class CustomerA {
		public static void main(String[] args) throws UnknownHostException, IOException {
			Client client=new Client();
			client.creaSockeet();
			StringBuilder builder=new StringBuilder("code:200;"); //消息标示
			builder.append("content:你好吗?");
			client.pub(builder.toString());
		}
}

在去启动我们的客户A，这个时候Server控制台会打印。

已经生产该消息code:200;content:你好吗?/t【现仓储量为】:1
code:200;content:你好吗?
然后再创建我们的消费者客户B，这个时候客户B的控制台会打印

public class CustomerB {
		public static void main(String[] args) throws UnknownHostException, IOException {
			Client client=new Client();
			client.creaSockeet();
			StringBuilder builder=new StringBuilder("code:400"); //消息标示
			String message=client.sub(builder.toString());
			System.out.println("获取的消息为:"+message);//获取消息
		}
}

获取的消息为:code:200;content:你好吗?

而我们的Server控制台会打印

【已经消费该消息】:code:200;content:你好吗?/t【现仓储量为】:0

时间： 2024-10-05 07:15:01

自己动手实现消息队列之JMS的相关文章

Java消息队列--JMS概述

1.什么是JMS JMS即Java消息服务(Java Message Service)应用程序接口,是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信.Java消息服务是一个与具体平台无关的API,绝大多数MOM提供商都对JMS提供支持(百度百科给出的概述).我们可以简单的理解:两个应用程序之间需要进行通信,我们使用一个JMS服务,进行中间的转发,通过JMS 的使用,我们可以解除两个程序之间的耦合. 2.JMS的优势 Async

JMS消息队列ActiveMQ(点对点模式)

生产者(producer)->消息队列(message queue) package com.java1234.activemq; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageProducer; import javax.jms.Session

JMS消息队列ActiveMQ(发布／订阅模式)

消费者1(Consumer)--订阅(subcribe)-->主题(Topic) package com.java1234.activemq2; import javax.jms.Connection; import javax.jms.ConnectionFactory; import javax.jms.Destination; import javax.jms.JMSException; import javax.jms.MessageConsumer; import javax.jms.

Flume 读取JMS 消息队列消息，并将消息写入HDFS

利用Apache Flume 读取JMS 消息队列消息.并将消息写入HDFS,flume agent配置例如以下: flume-agent.conf #name the  components on this agent agentHdfs.sources  = jms_source agentHdfs.sinks =  hdfs_sink agentHdfs.channels  = mem_channel #  Describe/configure the source agentHdfs.s

java JMS消息队列

http://blog.csdn.net/shirdrn/article/details/6362792 http://haohaoxuexi.iteye.com/blog/1893038 http://afreon.blog.163.com/blog/static/223794094201431422654237/ http://www.cnblogs.com/huang0925/p/3558690.html ActiveMQ第二弹:使用Spring JMS与ActiveMQ通讯 本文章的完整

Java消息队列ActiveMQ (一)--JMS基本概念

摘要:The Java Message Service (JMS) API is a messaging standard that allows application components based on the Java Platform Enterprise Edition (Java EE) to create, send, receive, and read messages. It enables distributed communication that is loosely

JMS与消息队列

JMS,Java Message Service,即Java消息服务. MOM,Message Oriented Miiddleware的英文缩写,指的是利用高效可靠的消息传递机制进行平台无关的数据交流,并基于数据通信来进行分布式系统的集成. JMS是Java的一套API标准,最初的目的是为了使应用程序能够访问现有的MOM系统,后来被许多现有的MOM供应商采用,并实现为MOM系统. 常见的MOM系统有Apache的ActiveMQ.BEA的RabbitMQ.阿里巴巴的RocketMQ.IBM的M

MQ消息队列（2）—— Java消息服务接口（JMS）

一.理解JMS   1.什么是JMS?         JMS即Java消息服务(Java Message Service)应用程序接口,API是一个消息服务的标准或者说是规范,允许应用程序组件基于JavaEE平台创建.发送.接收和读取消息.它使分布式通信耦合度更低,消息服务更加可靠以及异步性. 我们可以简单的理解:两个应用程序之间需要进行通信,我们使用一个JMS服务,进行中间的转发,通过JMS 的使用,我们可以解除两个程序之间的耦合. JMS不是消息队列,更不是某种消息队列协议.JMS是Jav

AMQP消息队列之RabbitMQ简单示例

前面一篇文章讲了如何快速搭建一个ActiveMQ的示例程序,ActiveMQ是JMS的实现,那这篇文章就再看下另外一种消息队列AMQP的代表实现RabbitMQ的简单示例吧.在具体讲解之前,先通过一个图来概览下: 1.添加Maven依赖 <!-- rabbitmq begin --> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit