首页 > 代码库 > 自己动手实现消息队列之JMS
自己动手实现消息队列之JMS
什么是JMS?JMS的诞生史?
在JMS还没有诞生前，每个企业都会有自己的一套内部消息系统，比如项目组A需要调用到项目组B的系统，项目组B也有可能会调用到项目组C的系统。这样每个公司都有自己的一套实现。很不规范，所以Apache基金会，为企业消息产品专门定义了一套规范。我们可以把JMS当作是一系列接口及相关语义的集合，通过这些接口和语义定义了JSM客户端如何去访问消息系统。简单点来说就是JMS主要干了两件事，定义通用的消息格式，和消息传递的模式。
体系结构
对象模型
下面将简单模拟一下一个基于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)
P2P
- P2P模式图
- 涉及到的概念
- 消息队列（Queue）
- 发送者(Sender)
- 接收者(Receiver)
- 每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。
P2P的特点
- 每个消息只有一个消费者（Consumer）(即一旦被消费，消息就不再在消息队列中)
- 发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列
- 接收者在成功接收消息之后需向队列应答成功
如果你希望发送的每个消息都应该被成功处理的话，那么你需要P2P模式。
- P2P模式图
Pub/Sub
- Pub/Sub模式图
- 涉及到的概念
- 主题（Topic）
- 发布者（Publisher）
- 订阅者（Subscriber）
客户端将消息发送到主题。多个发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。
Pub/Sub的特点
- 每个消息可以有多个消费者
- 发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息，而且为了消费消息，订阅者必须保持运行的状态。
- 为了缓和这样严格的时间相关性，JMS允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。
如果你希望发送的消息可以不被做任何处理、或者被一个消息者处理、或者可以被多个消费者处理的话，那么可以采用Pub/Sub模型
- 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
自己动手实现消息队列之JMS