JMS学习-JAVA简单实现生产者、消费者的P2P(point-to-point)模型

作者: admin 分类: JAVA,设计思路 发布时间: 2019-12-09 17:55  阅读: 159 views

最近学习知识点,频繁出现消息机制相关字眼。之前只是用过一些消息队列框架,如activeMQ,rocketMQ等。也只是基于使用,偶然来了兴趣想了解下jms功能的最简化实现是什么,这里做个整理。

JMS什么?

JMS即Java消息服务(Java Message Service)应用程序接口,是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。所以它只是一系列的约束/规范化的接口。

这里下载看的是 jms1.1 地址放出:https://mvnrepository.com/artifact/javax.jms/jms/1.1

这里只整理下JMS支持的五种消息类型

1.TextMessage(文本消息)
2.MapMessage(映射消息)
3.BytesMessage(字节消息)
4.StreamMessage(流消息)
5.ObjectMessage(对象消息)

其他的内容还包括像工厂、Session、Exception、Security等接口,相关信息请自行查看源码。

JAVA最简化实现点对点模式

点对点模式

代码包含五个部分:

队列的相关代码
  1. 消息队列(用于存储生产的消息 Queue)
  2. 消息服务端(用于接收客户端发来的消息进行处理)
  3. 消息客户端(提供发布、订阅的方法)
使用队列方的相关代码
  1. 生产者(利用客户端生产消息)
  2. 消费者(利用客户端消费消息)
  1. 消息队列相关代码
package com.chl.jms;

import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * 消息机制的存放、读取实现。
 * 这里使用ArrayBlockingQueue来实现队列模型(最简化的话,可以使用vector、stack等集合对象,但是要考虑数据的存取方式【先进先出、现金后出等方式】,还要考虑多线程情况下的线程安全问题)
 * 
 * 该类有两个方法
 * 一个是往队列存储消息
 * 一个是从队列获取消息
 * 注:这里增加了同步锁,保证了数据的正确性。
 * 
 * @author chenhailong
 *
 */
public class MessageQueue {

    /** 这里的队列对象视情况使用,可以使用list\queue\map类的集合对象 */
    private static Queue<String> msgQueue = new ArrayBlockingQueue<String>(100);

    /**
     * 生产消息
     * 
     * @param msg
     */
    public static void produce(String msg) {
        // 这里锁msgQueue,保证同时只有一个线程操作msgQueue
        synchronized (msgQueue) {
            while (msgQueue.size() >= 100) {
                // 这里对 queue的大小做判断,当容量不足,线程等待
                try {
                    msgQueue.wait();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
            msgQueue.add(msg);
            System.out.println("当前队列的大小为" + msgQueue.size());
            // 生产消息后,唤醒其他等待线程
            msgQueue.notifyAll();
        }
    }

    /**
     * 消费消息
     */
    public static String consumer() {
        synchronized (msgQueue) {
            while (msgQueue.size() < 0) {
                // 这里对 queue的大小做判断,当小于0,不执行消费任务
                try {
                    msgQueue.wait();
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
            // 获取并删除队列中的第一个元素
            String msg = msgQueue.poll();
            System.out.println("当前队列的大小为" + msgQueue.size());
            // 消费消息后,唤醒其他等待线程
            msgQueue.notifyAll();
            return msg;
        }
    }
}


之前的tomcat默认连接器多线程处理,也有object.wait()\notifyAll()等类似用法
tomcat的servlet多线程处理

  1. 消息服务端
package com.chl.jms;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * 消息机制的服务端实现
 * 服务端集成了thread类,增加了吞吐量。每一次请求都会开启一个新的线程。
 * 注:这里根据客户端请求的关键字进行判断,处理不同的业务|事件
 * @author chenhailong
 *
 */
public class QueueServer extends Thread {

    //构造函数,初始化socket对象
    QueueServer(Socket socket){
        this.socket = socket;
    }
    //独立线程的socket对象
    private Socket socket;

    public Socket getSocket() {
        return socket;
    }

    public void setSocket(Socket socket) {
        this.socket = socket;
    }


    @Override
    public void run() {
        //获取信息流
        try {
            BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));  
            PrintWriter out = new PrintWriter(socket.getOutputStream());  
            while(true) {
                String msg = in.readLine();
                if(msg != null) {
                    System.out.println("接收到的消息为:" + msg);
                    if(msg.equals("begin-product")) {
                        //如果是生产消息,则创建
                        MessageQueue.produce(msg);
                    }
                    if(msg.equals("consumer")) {
                        //如果是消费消息,则输出
                        out.println(MessageQueue.consumer());  //如果这里用 out.print, client端将收不到信息。
                        out.flush();
                    }
                    if(msg.equals("error")) {
                        break;
                    }
                }
            }

        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        //如果没有消息,则关闭退出
        try {
            socket.close();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

    @SuppressWarnings("resource")
    public static void main(String[] args) {
        try {
            ServerSocket ss = new ServerSocket(9999);
            System.out.println("JMS-server启动");
            while(true) {
                //多线程处理生产、消费
                QueueServer server = new QueueServer(ss.accept());
                server.start();
            }
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}


  1. 消息客户端
package com.chl.jms;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.net.InetAddress;
import java.net.Socket;
import java.net.UnknownHostException;

/**
 * 消息机制的客户端
 * 
 * 可以创建于服务器的socket连接。并提供了pub\sub两个方法
 * 
 * 
 * @author chenhailong
 *
 */
public class QueueClient {

     Socket server;

    //创建于服务端的连接
    public  void cs() throws UnknownHostException, IOException {
        server = new Socket(InetAddress.getLocalHost(),9999);
    }

    /**
     * 发布消息
     * @param msg 消息体
     * @throws IOException 
     */
    public  void pub(String msg) throws IOException {
//      BufferedReader in = new BufferedReader(new InputStreamReader(    
//                server.getInputStream()));    
        PrintWriter out = new PrintWriter(server.getOutputStream());    
            out.println(msg);  
            out.flush();     
        server.close();    
    }

    /**
     * 订阅消息
     * @param action 触发类型
     * @return 
     * @throws IOException
     */
    public  String sub(String action) throws IOException {
        PrintWriter out = new PrintWriter(server.getOutputStream());    
            out.println(action);  
            out.flush();     
            BufferedReader in = new BufferedReader(new InputStreamReader(    
                    server.getInputStream()));  

          String str = "";

//            char[] ret = new char[8096];
//            int len = 0;
//            while((len=in.read(ret, 0, ret.length))!=-1)
//            {
//                str = new String(ret,0,len);
//            }

           str = in.readLine(); //readLine方法某些情况会阻塞线程(如果信息流中没有回车或换行符,就会阻塞、所以要考虑使用环境)

            System.out.println("获取的消息为:"+str);  
        return str;  
    }
}

4.生产者

package com.chl.jms;

/**
 * 消费端的生产实现
 * @author chenhailong
 *
 */
public class CustomerA {

    public static void main(String[] args) throws Exception {
        QueueClient qc = new QueueClient();
        qc.cs();
        StringBuilder builder = new StringBuilder("begin-product");
        qc.pub(builder.toString());
        System.out.println("pub消息");
    }

}

5.消费者

package com.chl.jms;

/**
 * 消费端的订阅实现
 * @author chenhailong
 *
 */
public class CustomerB {

    public static void main(String[] args) throws Exception {
        QueueClient qc = new QueueClient();
        qc.cs();
        StringBuilder builder = new StringBuilder("consumer");
        String msg = qc.sub(builder.toString());
        System.out.println("获取的消息为:"+ msg);

    }

}

当然这不是最简化的代码,可以根据要求重新调整、删除扩展代码。通过这里的学习,可以比较清楚的了解到消息队列相关框架的最简单的实现,各种MQ又基于此进行了各种扩充。之后可以对其中的IO处理、模式设计等的优化、升级进行学习。


   原创文章,转载请标明本文链接: JMS学习-JAVA简单实现生产者、消费者的P2P(point-to-point)模型

如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!

发表评论

电子邮件地址不会被公开。 必填项已用*标注

更多阅读