文档章节

分布式Java--基于消息方式实现系统间通信

akane_oimo
 akane_oimo
发布于 09/14 23:22
字数 1881
阅读 16
收藏 15

分布式系统之间通信可以分为两种:

  • 基于消息方式实现系统间通信
  • 基于远程调用方式实现系统间通信

基于消息方式实现系统间通信

分布式子系统之间需要通信时,就发送消息。一般通信的两个要点是:消息处理和消息传输。

  • 消息处理:例如读取数据和写入数据。基于消息方式实现系统通信的消息处理可以分为同步消息和异步消息。同步消息一般采用的是BIO(Blocking IO)和NIO(Non-Blocking IO);异步消息一般采用AIO方式。
  • 消息传输:消息传输需要借助网络协议来实现,TCP/IP协议和UDP/IP协议可以用来完成消息传输。

术语解释:

  1. BIO:同步阻塞IO。就是当发生IO的读或者写操作时,均为阻塞操作。只有程序读到了流或者将流写入操作系统后,才会释放资源。
  2. NIO: 同步非阻塞IO。是基于事件驱动思想的。从程序角度想,当发起IO的读和写操作时,是非阻塞的。当Socket有流可读或者可以写Socket时,操作系统会通知应用程序进行处理,应用再将流读取到缓冲区或操作系统。
  3. AIO: 异步IO。同样基于事件驱动思想。当有流可读取时,操作系统会将流读取到read方法的缓冲区,然后通知应用程序;对于写操作,操作系统将write方法传入的流写入完毕时,操作系统主动通知应用程序。
  4. TCP/IP: 一种可靠的网络数据传输协议。要求通信双方先建立连接,再进行通信。
  5. UDP/IP: 一种不可靠的网络数据传输协议。并不直接给通信双方建立连接,而是发送到网络上通信。

四种方法实现基于消息进行系统间通信

TCP/IP+BIO

在Java中可基于Socket、ServerSocket来实现TCP/IP+BIO的系统通信。

  • Socket主要用于实现建立连接即网络IO的操作
  • ServerSocket主要用于实现服务器端口的监听即Socket对象的获取

为了满足服务端可以同时接受多个请求,最简单的方法是生成多个Socket。但这样会产生两个问题:

  • 生成太对Socket会消耗过多资源
  • 频繁创建Socket会导致系统性能的不足

为了解决上面的问题,通常采用连接池的方式来维护Socket。一方面能限制Socket的个数;另一方面避免重复创建Socket带来的性能下降问题。这里有一个问题就是设置合适的相应超时时间。因为连接池中Socket个数是有限的,肯定会造成激烈的竞争和等待。

客户端代码:

//创建连接
Socket socket = new Socket(目标IP或域名, 目标端口);
//BufferedReader用于读取服务端返回的数据
BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
//PrintWriter向服务器写入流
PrintWriter out = new PrintWriter(socket.getOutputStream(),true);
//像服务端发送流
out.println("hello");
//阻塞读取服务端的返回信息
in.readLine();

服务端代码:

//创建对本地端口的监听
PrintWriter out = new PrintWriter(socket.getOutputStream(),true);
//向服务器发送字符串信息
out.println("hello");
//阻塞读取服务端的返回信息
in.readLine();

TCP/IP+NIO

Java可以基于Clannel和Selector的相关类来实现TCP/IP+NIO方式的系统间通信。Channel有SocketClannel和ServerSocketChannel两种。

  • SocketClannel: 用于建立连接、监听事件及操作读写。
  • ServerSocketClannel: 用于监听端口即监听连接事件。
  • Selecter: 获取是否有要处理的事件。

客户端代码

SocketChannel channel = SocketChannel.open();
//设置为非阻塞模式
channel.configureBlocking(false);
//对于非阻塞模式,立即返回false,表示连接正在建立中
channel.connect(SocketAdress);
Selector selector = Selector.open();
//向channel注册selector以及感兴趣的连接事件
channel.regester(selector,SelectionKey.OP_CONNECT);
//阻塞至有感兴趣的IO事件发生,或到达超时时间
int nKeys = selector.select(超时时间【毫秒计】);
//如果希望一直等待知道有感兴趣的事件发生
//int nKeys = selector.select();
//如果希望不阻塞直接返回当前是否有感兴趣的事件发生
//int nKeys = selector.selectNow();

//如果有感兴趣的事件
SelectionKey sKey = null;
if(nKeys>0){
    Set<SelectionKey> keys = selector.selectedKeys();
    for(SelectionKey key:keys){
        //对于发生连接的事件
        if(key.isConnectable()){
            SocketChannel sc = (SocketChannel)key.channel();
            sc.configureBlocking(false);
            //注册感兴趣的IO读事件
            sKey = sc.register(selector,SelectionKey.OP_READ);
            //完成连接的建立
            sc.finishConnect();
        }
        //有流可读取
        else if(key.isReadable()){
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            SocketChannel sc = (SocketChannel) key.channel();
            int readBytes = 0;
            try{
                int ret = 0;
                try{
                    //读取目前可读取的值,此步为阻塞操作
                    while((ret=sc.read(buffer))>0){
                        readBytes += ret;
                    }
                }
                fanally{
                    buffer.flip();
                }
             }
             finally{
                 if(buffer!=null){
                        buffer.clear();
                 }
             }
        }
        //可写入流
        else if(key.isWritable()){
            //取消对OP_WRITE事件的注册
            key.interestOps(key.interestOps() & (!SelectionKey.OP_WRITE));
            SocketChannel sc = (SocketChannel) key.channel();
            //此步为阻塞操作
            int writtenedSize = sc.write(ByteBuffer);
            //如未写入,则继续注册感兴趣的OP_WRITE事件
            if(writtenedSize==0){
                key.interestOps(key.interestOps()|SelectionKey.OP_WRITE);
            }
        }
    }
    Selector.selectedKeys().clear();
}
//对于要写入的流,可直接调用channel.write来完成。只有在未写入成功时才要注册OP_WRITE事件
int wSize = channel.write(ByteBuffer);
if(wSize == 0){
    key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
}

服务端代码

ServerSocketChannel ssc = ServerSocketChannel.open();
ServerSocket serverSocket = ssc.socket();
//绑定要监听的接口
serverSocket.bind(new InetSocketAdress(port));
ssc.configureBlocking(false);
//注册感兴趣的连接建立事件
ssc.register(selector,SelectionKey.OP_ACCEPT);

UDP/IP+BIO

Java对UDP/IP方式的网络数据传输同样采用Socket机制,只是UDP/IP下的Socket没有建立连接,因此无法双向通信。如果需要双向通信,必须两端都生成UDP Server。

Java中通过DatagramSocket和DatagramPacket来实现UDP/IP+BIO方式和系统间通信。

  • DatagramSocket:负责监听端口和读写数据
  • DatagramPacket:作为数据流对象进行传输

由于UDP双端不建立连接,所以也就不存在竞争问题,只是最终读写流的动作是同步的。

关键代码(服务端和客户端基本一样)

//如果希望双向通信,必须启动一个监听端口承担服务器的职责
//如果不能绑定到指定端口,则抛出SocketException
DatagramSocket serverSocket = new DatagramSocket(监听的端口);
byte[] buffer = new byte[65507];
DatagramPacket receivePacket = new DatagramPacket(buffer,buffer.length);
DatagramSocket socket = new DatagramSocket();
DatagramPacket packet = new DatagramPacket(datas,datas.length,server.length);
//阻塞方式发送packet到指定的服务器和端口
socket.send(packet);
//阻塞并同步读取流消息,如果读取的流消息比packet长,则删除更长的消息
//当连接不上目标地址和端口时,抛出PortUnreachableException
DatagramSocket.setSoTimeout(超时时间--毫秒级);
serverSocket.receive(receivePacket);

UDP/IP+NIO

Java中可以通过DatagramClannel和ByteBuffer来实现UDP/IP方式的系统间通信。

  • DatagramClannel:负责监听端口及进行读写
  • ByteBuffer:用于数据传输

关键代码(客户端和服务端都类似)

//读取流信息
DatagramChannel receiveChannel = DatagramChannel.open();
receiveChannel.configureBlocking(false);
DatagramSocket socket = receiveChannel.socket();
socket.bind(new InetSocketAddress(rport));
Selector selector = Selector.open();
receiveChannel.register(selector, SelectionKey.OP_REEAD);
//之后即可像TCP/IP+NIO中对selector遍历一样的方式进行流信息的读取
//...


//写入流信息
DatagramChannel sendChannel = DatagramChannel.open();
sendChannel.configureBlocking(false);
SocketAdress target = new InetSocketAdress("127.0.0.1",sport);
sendChannel.connect(target);
//阻塞写入流
sendChannel.write(ByteBuffer);

 

© 著作权归作者所有

共有 人打赏支持
akane_oimo
粉丝 19
博文 129
码字总数 155529
作品 0
南京
其他
分布式服务架构之java远程调用技术浅析

在 分布式服务框架中,一个最基础的问题就是远程服务是怎么通讯的,在Java领域中有很多可实现远程通讯的技术,例如:RMI、MINA、ESB、 Burlap、Hessian、SOAP、EJB和JMS等,这些名词之间到底...

aldo
2012/11/20
0
2
分布式服务架构之java远程调用技术浅析

在分布式服务框架中,一个最基础的问题就是远程服务是怎么通讯的,在Java领域中有很多可实现远程通讯的技术,例如:RMI、MINA、ESB、Burlap、Hessian、SOAP、EJB和JMS等,这些名词之间到底是...

weyling
2013/12/28
0
0
Java远程通讯可选技术及原理

原文网址链接http://www.blogjava.net/BlueDavy/archive/2008/03/04/182077.html 在分布式服务框架中,一个最基础的问题就是远程服务是怎么通讯的,在Java领域中有很多可实现远程通讯的技术,...

数通畅联
2015/02/02
0
0
分布式Java--基于远程调用实现系统间通信

分布式系统之间通信可以分为两种: 基于消息方式实现系统间通信 基于远程调用方式实现系统间通信 基于远程调用实现系统间通信 远程调用方式就是尽可能将系统间的调用模拟为系统内的调用,让使...

akane_oimo
09/15
0
0
Iot的七大通信协议,你了解几个

  通信对物联网来说十分常用且关键,无论是近距离无线传输技术还是移动通信技术,都影响着物联网的发展。而在通信中,通信协议尤其重要,是双方实体完成通信或服务所必须遵循的规则和约定。...

昇润科技
05/08
0
0

没有更多内容

加载失败,请刷新页面

加载更多

xilinx资源

本系列教学视频由赛灵思高级战略应用工程师带领你:从零开始,一步步深入 掌握 HLS 以及 UltraFAST 设计方法,帮助您成为系统设计和算法加速的大拿! http://www.eetrend.com/topics/2018-0...

whoisliang
23分钟前
0
0
=====BJmeter性能测试小接=====

一、性能测试分类 1、负载测试: 通过逐步加压的方法,达到既定的性能阈值的目标,阈值的设定应是小于某个值,如cpu使用率小于等于80% 2、压力测试: 通过逐步加压的方法,使得系统的某些资源...

覃光林
26分钟前
1
0
企业级开源四层负载均衡解决方案--LVS

网盘链接 企业级开源四层负载均衡解决方案--LVS 本课程将在Linux环境下,学习配置使用LVS,对Web集群和MySQL集群进行负载均衡,并结合利用Keepalived实现负载均衡器的高可用,实现对后端Rea...

qq__2304636824
32分钟前
2
0
Windows上安装Spacemacs

emacs安装 下载地址emacs 安装比较简单,解压后执行\bin\addpm.exe即可 emacs配置 emacs的默认配置文件路径和.emacs.d文件夹都是在Windows主目录下的 C:\Users\Administrator\AppData\Roami...

yxmsw2007
48分钟前
0
0
OSChina 周一乱弹 —— 鱼生不值得

Osc乱弹歌单(2018)请戳(这里) 【今日歌曲】 @瘟神灬念:分享新裤子的单曲《没有理想的人不伤心 (Remix版)》: 《没有理想的人不伤心 (Remix版)》- 新裤子 手机党少年们想听歌,请使劲儿戳...

小小编辑
今天
215
9

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部