java分散式系統通訊框架

阿新 • • 發佈：2019-02-11

1.Jboss Netty專案-優秀的NIO框架
下載[url]http://www.jboss.org/netty/downloads.html[/url]
簡介[url]http://www.jboss.org/netty[/url]
Jboss的Netty專案旨在提供非同步驅動的網路應用程式的框架和工具, 可以用於快速開發, 可方便維護的、高效能的、高擴充套件性的伺服器/客戶端之間的通迅協議。
Plurk是一個快速成長，並且已經成為網際網路上最大的幾個使用python語言的網站之之一，Plurk嘗試了以下一些框架/web容器：
Python Twisted：太耗資源了
Jetty：iGoogle的web容器，可惜一萬併發的時候用了2G的記憶體
Apache Tomcat：太吃記憶體了
Apache MINA：文件很糟糕，可擴充套件性很差
最後，Plurk發現Netty非常優秀，在一臺四核的伺服器上，10萬的併發只用了some GB的記憶體和20%左右的CPU
JBoss社群最近釋出了Netty 3.1.0，為使用者提供了編寫客戶/服務網路應用的另一選擇。Netty自稱是：

一款非同步的事件驅動的網路應用框架和工具，用於快速開發可維護的高效能、高擴充套件性協議伺服器和客戶端。也就是說，Netty是一個NIO客戶端/伺服器框架，支援快速、簡單地開發網路應用，如協議伺服器和客戶端。它極大簡化了網路程式設計，如TCP和UDP套接字伺服器。

關於Netty的分類，它與Apache Mina和Grizzly屬於同一舞臺。該最新版本包含了一系列功能和效能、API可用性的增強。其中包括：

* 更簡單的大資料流（例子）
* 更多可靠的OutOfMemoryError預防機制
* 新的傳輸
o 基於OIO和NIO的UDP傳輸
o VM內傳輸
o HTTP隧道
* 與Google Protocol Buffers整合
* 與JBoss Microcontainer、OSGI、Guice和Spring整合

HTTP隧道功能滿足了使用者的迫切需求，在早期釋出說明中總結到：

HTTP隧道傳輸（位於org.jboss.netty.channel.socket.http）是一種套接字傳輸，支援任意已經存在的套接字應用在HTTP之上通過代理傳輸。這種傳輸在需要越過防火牆而不修改現存伺服器應用時特別有用。工作原理如下：

Http隧道客戶端套接字通道 --> 對HTTP友好的防火牆 --> Servlet容器 (如Tomcat、Jetty) --> Http隧道Servlet --> 你的伺服器應用

當然在選擇一個框架用於開發網路應用時，效能和可擴充套件性是重要因素。Netty團隊在他們的網站上提供了效能資訊。你也可以在Apache Mina網站上找到它的效能資料。Nicholas Hagen釋出了一系列博文，記錄了他如何選擇 Mina、Grizzly或者Netty。最終他決定選用Netty，不過指出：

總的來說，在效能、記憶體和功能使用方面，我傾向於Netty，而不是Mina和Grizzly。請注意你需要通過自己的分析來決定哪種框架適合你的需求。

Netty下一版本3.2.0，目前正在開發過程中。

JBoss社群最近釋出了Netty 3.1.0，為使用者提供了編寫客戶/服務網路應用的另一選擇。Netty自稱是：

一款非同步的事件驅動的網路應用框架和工具，用於快速開發可維護的高效能、高擴充套件性協議伺服器和客戶端。也就是說，Netty是一個NIO客戶端/伺服器框架，支援快速、簡單地開發網路應用，如協議伺服器和客戶端。它極大簡化了網路程式設計，如TCP和UDP套接字伺服器。

關於Netty的分類，它與Apache Mina和Grizzly屬於同一舞臺。該最新版本包含了一系列功能和效能、API可用性的增強。其中包括：

* 更簡單的大資料流（例子）
* 更多可靠的OutOfMemoryError預防機制
* 新的傳輸
o 基於OIO和NIO的UDP傳輸
o VM內傳輸
o HTTP隧道
* 與Google Protocol Buffers整合
* 與JBoss Microcontainer、OSGI、Guice和Spring整合

HTTP隧道功能滿足了使用者的迫切需求，在早期釋出說明中總結到：

HTTP隧道傳輸（位於org.jboss.netty.channel.socket.http）是一種套接字傳輸，支援任意已經存在的套接字應用在HTTP之上通過代理傳輸。這種傳輸在需要越過防火牆而不修改現存伺服器應用時特別有用。工作原理如下：

Http隧道客戶端套接字通道 --> 對HTTP友好的防火牆 --> Servlet容器 (如Tomcat、Jetty) --> Http隧道Servlet --> 你的伺服器應用

當然在選擇一個框架用於開發網路應用時，效能和可擴充套件性是重要因素。Netty團隊在他們的網站上提供了效能資訊。你也可以在Apache Mina網站上找到它的效能資料。Nicholas Hagen釋出了一系列博文，記錄了他如何選擇 Mina、Grizzly或者Netty。最終他決定選用Netty，不過指出：

總的來說，在效能、記憶體和功能使用方面，我傾向於Netty，而不是Mina和Grizzly。請注意你需要通過自己的分析來決定哪種框架適合你的需求。

Netty下一版本3.2.0，目前正在開發過程中，路線圖可以在這裡檢視。
[img]http://www.iteye.com/upload/attachment/130154/cb5e82c1-4f14-3fbc-a6b3-43d6b5cba142.png[/img]

此版本的主要更新：
處理大容量資料流更簡單
處理協議編碼和單元測試更簡單
I/O超時和idle狀態檢測
應用程式的關閉更簡單，更安全
更可靠的OutOfMemoryError預防
新的傳輸方式：
1.基於OIO和NIO的UDP傳輸
2.本地傳輸(又名 in-VM傳輸)
3.HTTP通道，可繞過防火牆
新的編碼器：
1.HTTP客戶端和伺服器端
2.用於實現各種專有協議的工具
與其他技術的整合：
1.Google Protocol Buffers
2.JBoss Microcontainer, OSGi, Guice以及Spring

package example.helloword.server;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;

import example.helloword.NetConstant;

public class Server
{
    private static Server server = new Server();

    private ServerBootstrap bootstrap;

    private Server()
    {}

    public static Server getInstance()
    {
        return server;
    }

    public void start()
    {
        bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
                Executors.newCachedThreadPool(), Executors
                        .newCachedThreadPool()));
        bootstrap.setPipelineFactory(new ServerPipelineFactory());
        bootstrap.bind(new InetSocketAddress(NetConstant.server_port));
    }

    public void stop()
    {
        bootstrap.releaseExternalResources();
    }

    public static void main(String[] args)
    {
        Server server = Server.getInstance();
        server.start();
    }
}

package example.helloword.server;

import static org.jboss.netty.channel.Channels.pipeline;

import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;

public class ServerPipelineFactory implements ChannelPipelineFactory
{
    public ChannelPipeline getPipeline() throws Exception
    {
        ChannelPipeline pipleline = pipeline();
        pipleline.addLast("encode", new StringEncoder());
        pipleline.addLast("decode", new StringDecoder());
        pipleline.addLast("handler", new ServerHandler());
        return pipleline;
    }
}

package example.helloword.server;

import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;

public class ServerHandler extends SimpleChannelUpstreamHandler
{
    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
            throws Exception
    {
        System.out.println("recive message,message content:" + e.getMessage());
        e.getChannel().write("byte");

    }

    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
            throws Exception
    {
        e.getChannel().close();
    }
}

package example.helloword.client22;

import static org.jboss.netty.channel.Channels.pipeline;

import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
import org.jboss.netty.handler.codec.frame.LengthFieldPrepender;
import org.jboss.netty.handler.codec.string.StringDecoder;
import org.jboss.netty.handler.codec.string.StringEncoder;

public class ClientPipelineFactory implements ChannelPipelineFactory
{
    public ChannelPipeline getPipeline() throws Exception
    {
        ChannelPipeline pipleline = pipeline();  
        pipleline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));  
        pipleline.addLast("frameEncode", new LengthFieldPrepender(4, false));
        pipleline.addLast("encode", new StringEncoder());  
        pipleline.addLast("decode", new StringDecoder());  
        pipleline.addLast("handler", new ClinetHandler());  
        return pipleline;  
    }
}

package example.helloword.client22;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

import example.helloword.NetConstant;
import example.helloword.client2.ClientPipelineFactory;

public class ClientPool
{
    public static ClientPool clientPool = new ClientPool();

    private ClientBootstrap bootstrap;

    private ClientPool()
    {
        bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(
                Executors.newCachedThreadPool(), Executors
                        .newCachedThreadPool()));

        bootstrap.setPipelineFactory(new ClientPipelineFactory());
        bootstrap.setOption("tcpNoDelay", true);
        bootstrap.setOption("keepAlive", true);
    }

    public static ClientPool getInstance()
    {
        return clientPool;
    }

    public void getChannelFuture(String host, int port, String message)
    {
        ChannelFuture future = bootstrap.connect(new InetSocketAddress(host,
                NetConstant.server_port));
        future.awaitUninterruptibly();
        if (!future.isSuccess())
        {
            future.getCause().printStackTrace();
            future.getChannel().getCloseFuture().awaitUninterruptibly(); 
            return;
        }
        future.getChannel().write(message);
    }

    public static void main(String[] args) throws InterruptedException
    {
        for (int i = 0; i < 1000; i++)
        {
            ClientPool.getInstance().getChannelFuture("127.0.0.1", 0,
                    "test" + i);
            Thread.sleep(1000 * 3);
        }
    }
}

2、apache mina
下載[url]http://mina.apache.org/downloads.html[/url]

public class Test {
	public static void main(String[] args) throws IOException {
		HttpServer httpServer;
		httpServer = HttpServer.create(new InetSocketAddress(81), 5);
		httpServer.createContext("/", new Handler());
		httpServer.start();
	}

	static class Handler implements HttpHandler {
		public void handle(HttpExchange exchange) throws IOException {
			Headers requestHeaders = exchange.getRequestHeaders();
			Headers responseHeaders = exchange.getResponseHeaders();
			responseHeaders.set("Content-Type", "text/plain");
			exchange.sendResponseHeaders(200, 0L);
			OutputStream responseBody = new BufferedOutputStream(exchange.getResponseBody(), 64*1024);
			responseBody.write("Hello!".getBytes());
			responseBody.close();
			exchange.close();
		}
	}
}

import java.net.InetSocketAddress;

import org.apache.mina.core.filterchain.DefaultIoFilterChainBuilder;
import org.apache.mina.example.echoserver.ssl.BogusSslContextFactory;
import org.apache.mina.filter.ssl.SslFilter;
import org.apache.mina.transport.socket.SocketAcceptor;
import org.apache.mina.transport.socket.nio.NioSocketAcceptor;

/**
 * (<b>Entry point</b>) Echo server
 *
 * @author The Apache MINA Project ([email protected])
 * @version $Rev: 677923 $, $Date: 2008-07-18 16:55:24 +0200 (Fri, 18 Jul 2008) $
 */
public class Main {
    /** Choose your favorite port number. */
    private static final int PORT = 8080;

    /** Set this to true if you want to make the server SSL */
    private static final boolean USE_SSL = false;

    public static void main(String[] args) throws Exception {
        SocketAcceptor acceptor = new NioSocketAcceptor();
        DefaultIoFilterChainBuilder chain = acceptor.getFilterChain();


        // Bind
        acceptor.setHandler(new EchoProtocolHandler());
        acceptor.bind(new InetSocketAddress(PORT));

        System.out.println("Listening on port " + PORT);

        for (;;) {
            System.out.println("R: " + acceptor.getStatistics().getReadBytesThroughput() + 
                ", W: " + acceptor.getStatistics().getWrittenBytesThroughput());
            Thread.sleep(3000);
        }
    }

 }

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.ssl.SslFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * {@link IoHandler} implementation for echo server.
 *
 * @author The Apache MINA Project ([email protected])
 * @version $Rev: 713957 $, $Date: 2008-11-14 10:27:16 +0100 (Fri, 14 Nov 2008) $,
 */
public class EchoProtocolHandler extends IoHandlerAdapter {
    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    public void sessionCreated(IoSession session) {
        session.getConfig().setIdleTime(IdleStatus.BOTH_IDLE, 10);

        // We're going to use SSL negotiation notification.
        session.setAttribute(SslFilter.USE_NOTIFICATION);
    }

    @Override
    public void sessionClosed(IoSession session) throws Exception {
        logger.info("CLOSED");
    }

    @Override
    public void sessionOpened(IoSession session) throws Exception {
        logger.info("OPENED");
    }

    @Override
    public void sessionIdle(IoSession session, IdleStatus status) {
        logger.info("*** IDLE #" + session.getIdleCount(IdleStatus.BOTH_IDLE) + " ***");
    }

    @Override
    public void exceptionCaught(IoSession session, Throwable cause) {
        session.close(true);
    }

    @Override
    public void messageReceived(IoSession session, Object message)
            throws Exception {
        // Write the received data back to remote peer
        session.write(((IoBuffer) message).duplicate());
    }
}

public class Test {
	public static void main(String[] args) throws IOException {
		HttpServer httpServer;
		httpServer = HttpServer.create(new InetSocketAddress(81), 5);
		httpServer.createContext("/", new Handler());
		httpServer.start();
	}

	static class Handler implements HttpHandler {
		public void handle(HttpExchange exchange) throws IOException {
			Headers requestHeaders = exchange.getRequestHeaders();
			Headers responseHeaders = exchange.getResponseHeaders();
			responseHeaders.set("Content-Type", "text/plain");
			exchange.sendResponseHeaders(200, 0L);
			OutputStream responseBody = new BufferedOutputStream(exchange.getResponseBody(), 64*1024);
			responseBody.write("Hello!".getBytes());
			responseBody.close();
			exchange.close();
		}
	}
}

java分散式系統通訊框架

java分散式系統通訊框架

ELK 實現 Java 分散式系統日誌分析架構

java分散式系統部署學習（九）ansible-playbook進階

Java分散式系統搭建--入門1

ELK(elasticsearch+logstash+kibana)實現Java分散式系統日誌分析架構

java分散式系統部署學習(二)ansible構架

java分散式系統部署學習（六）ansible Ad-hoc與commands模組

java分散式系統部署學習（五）ansible Dynamic Inventory

java分散式系統部署學習（四）ansible配置ansible.cfg

Java分散式系統高併發解決方案

JAVA NIO非同步通訊框架MINA選型和使用的幾個細節(概述入門，UDP, 心跳)

java分散式系統部署學習（三）ansible的Inventory與Patterns

iBase4J是Java的分散式系統架構使用Springboot整合開源框架

分散式系統詳解--框架（Hadoop--JAVA操作HDFS檔案）

Java架構師面試題全集：Java基礎+技術框架+系統架構+分散式系統

分散式通訊框架 - rmi

基於Java NIO2實現的非同步非阻塞訊息通訊框架

分散式系統詳解--基礎知識（通訊）

分散式系統詳解--框架（Hadoop-叢集搭建）

分散式系統詳解--框架（Hadoop-單機版搭建）

java分散式系統通訊框架

相關推薦