文档章节

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

akane_oimo
 akane_oimo
发布于 2018/09/14 23:22
字数 1881
阅读 36
收藏 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
粉丝 21
博文 132
码字总数 161923
作品 0
南京
其他
私信 提问
J2EE平台简介 

1.1.1 J2EE规范 J2EE(Java 2 Platform,Enterprise Edition)是SUN公司定义的一个开发分布式企业级应用的规范。它提供了一个多层次的分布式应用模型和一系列开发技术规范。多层次分布式应用模...

曾赛
2009/09/06
382
0
分布式服务架构之java远程调用技术浅析

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

weyling
2013/12/28
0
0
分布式Java--基于远程调用实现系统间通信

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

akane_oimo
2018/09/15
0
0
晒晒你看过的哪些Java相关的书。

@红薯 要我们晒Jar包 , 我倒想看看大家都看过什么Java编程相关的书籍,我先来:如下 1、iText in Action 主要内容: a、自动化静态和动态XFA表单 b、如何从XML生成动态PDF或数据库 c、如何添...

刘学炜
2012/09/27
317
4
分布式服务架构之java远程调用技术浅析

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

aldo
2012/11/20
0
2

没有更多内容

加载失败,请刷新页面

加载更多

漏洞防御与修复工作

漏洞管理工作是企业安全建设必不可少的一环,在风险管理工作中,漏洞管理能够防患于未然,企业对漏洞管理有着广泛的基础建设和实践经验。但随着攻防技术的发展,传统漏洞管理的安全技术和管理...

linuxprobe16
31分钟前
1
0
MicroPython技术及应用前景

1 Micropython技术是什么? MicroPython极精简高效的实现了Python3语言。它包含Python标准库的一小部分,能在单片机和受限环境中运行。 1.1 MicroPython发展 由剑桥大学的理论物理学家乔治....

bodasisiter
37分钟前
1
0
跟我学Spring Cloud(Finchley版)-13-通用方式使用Hystrix

本节详细讲解使用Hystrix的通用方式。 简介 Hystrix是由Netflix开源的一个延迟和容错库,用于隔离访问远程系统、服务或者第三方库,防止级联失败,从而提升系统的可用性与容错性。Hystrix主要...

周立_ITMuch
45分钟前
1
0
🛠️Hanjst/汉吉斯特更新加JavaScript运行时优化等

这是 Hanjst/汉吉斯特 发布以来的首个主要升级更新版本。这次的主要升级更新的内容包括移除HTML Comments注释行, 优化在 Hanjst include模板文件时的JavaScript运行时环境。 Hanjst 在设计和...

wadelau
今天
2
0
OSChina 周六乱弹 —— 舔狗是没有好下场的

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @我没有抓狂 :#今天听什么# #今天听这个# 分享 Nirvana 的歌曲《Smells Like Teen Spi...》 《Smells Like Teen Spi...》- Nirvana 手机党少...

小小编辑
今天
526
14

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部