2. 程式人生 > >Netty實現客戶端和服務端通信簡單例子

Netty實現客戶端和服務端通信簡單例子

阿新 • • 發佈：2018-03-24
啟動服務 ali tty 過程 等等 服務器初始化 讀寫操作 extends ask

Netty是建立在NIO基礎之上，Netty在NIO之上又提供了更高層次的抽象。

在Netty裏面，Accept連接可以使用單獨的線程池去處理，讀寫操作又是另外的線程池來處理。

Accept連接和讀寫操作也可以使用同一個線程池來進行處理。而請求處理邏輯既可以使用單獨的線程池進行處理，也可以跟放在讀寫線程一塊處理。線程池中的每一個線程都是NIO線程。用戶可以根據實際情況進行組裝，構造出滿足系統需求的並發模型。

Netty提供了內置的常用編解碼器，包括行編解碼器［一行一個請求］，前綴長度編解碼器［前N個字節定義請求的字節長度］，可重放解碼器［記錄半包消息的狀態］，HTTP編解碼器，WebSocket消息編解碼器等等

Netty提供了一些列生命周期回調接口，當一個完整的請求到達時，當一個連接關閉時，當一個連接建立時，用戶都會收到回調事件，然後進行邏輯處理。

Netty可以同時管理多個端口，可以使用NIO客戶端模型，這些對於RPC服務是很有必要的。

Netty除了可以處理TCP Socket之外，還可以處理UDP Socket。

在消息讀寫過程中，需要大量使用ByteBuffer，Netty對ByteBuffer在性能和使用的便捷性上都進行了優化和抽象。

代碼：

服務端：

package com.kinson.netty.server;

import io.netty.bootstrap.ServerBootstrap;

 
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

/**
 * descripiton:服務端
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 15:37
 * @modifier:
 * 
 
@since:
 */
public class NettyServer {
    /**
     * 端口
     */
    private int port;

    public NettyServer(int port) {
        this.port = port;
    }

    public void run() {
        //EventLoopGroup是用來處理IO操作的多線程事件循環器
        //負責接收客戶端連接線程
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        //負責處理客戶端i/o事件、task任務、監聽任務組
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        //啟動 NIO 服務的輔助啟動類
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup);
        //配置 Channel
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new ServerIniterHandler());
        //BACKLOG用於構造服務端套接字ServerSocket對象，
        // 標識當服務器請求處理線程全滿時，用於臨時存放已完成三次握手的請求的隊列的最大長度
        bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
        //是否啟用心跳保活機制
        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
        try {
            //綁定服務端口監聽
            Channel channel = bootstrap.bind(port).sync().channel();
            System.out.println("server run in port " + port);
            //服務器關閉監聽
            /*channel.closeFuture().sync()實際是如何工作:
            channel.closeFuture()不做任何操作，只是簡單的返回channel對象中的closeFuture對象，對於每個Channel對象，都會有唯一的一個CloseFuture，用來表示關閉的Future，
            所有執行channel.closeFuture().sync()就是執行的CloseFuturn的sync方法，從上面的解釋可以知道，這步是會將當前線程阻塞在CloseFuture上*/
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            //關閉事件流組
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) {
        new NettyServer(8899).run();
    }
}

服務端業務邏輯處理：

package com.kinson.netty.server;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

/**
 * descripiton: 服務器的處理邏輯
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 15:50
 * @modifier:
 * @since:
 */
public class ServerHandler extends SimpleChannelInboundHandler<String> {

    /**
     * 所有的活動用戶
     */
    public static final ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 讀取消息通道
     *
     * @param context
     * @param s
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext context, String s)
            throws Exception {
        Channel channel = context.channel();
        //當有用戶發送消息的時候，對其他的用戶發送消息
        for (Channel ch : group) {
            if (ch == channel) {
                ch.writeAndFlush("[you]: " + s + "\n");
            } else {
                ch.writeAndFlush("[" + channel.remoteAddress() + "]: " + s + "\n");
            }
        }
        System.out.println("[" + channel.remoteAddress() + "]: " + s + "\n");
    }

    /**
     * 處理新加的消息通道
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        for (Channel ch : group) {
            if (ch == channel) {
                ch.writeAndFlush("[" + channel.remoteAddress() + "] coming");
            }
        }
        group.add(channel);
    }

    /**
     * 處理退出消息通道
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        for (Channel ch : group) {
            if (ch == channel) {
                ch.writeAndFlush("[" + channel.remoteAddress() + "] leaving");
            }
        }
        group.remove(channel);
    }

    /**
     * 在建立連接時發送消息
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        boolean active = channel.isActive();
        if (active) {
            System.out.println("[" + channel.remoteAddress() + "] is online");
        } else {
            System.out.println("[" + channel.remoteAddress() + "] is offline");
        }
        ctx.writeAndFlush("[server]: welcome");
    }

    /**
     * 退出時發送消息
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        if (!channel.isActive()) {
            System.out.println("[" + channel.remoteAddress() + "] is offline");
        } else {
            System.out.println("[" + channel.remoteAddress() + "] is online");
        }
    }

    /**
     * 異常捕獲
     *
     * @param ctx
     * @param e
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) throws Exception {
        Channel channel = ctx.channel();
        System.out.println("[" + channel.remoteAddress() + "] leave the room");
        ctx.close().sync();
    }

}

服務端處理器註冊：

package com.kinson.netty.server;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;


/**
 * descripiton: 服務器初始化
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 15:46
 * @modifier:
 * @since:
 */
public class ServerIniterHandler extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        //管道註冊handler
        ChannelPipeline pipeline = socketChannel.pipeline();
        //編碼通道處理
        pipeline.addLast("decode", new StringDecoder());
        //轉碼通道處理
        pipeline.addLast("encode", new StringEncoder());
        //聊天服務通道處理
        pipeline.addLast("chat", new ServerHandler());
    }
}

客戶端：

package com.kinson.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.lang3.StringUtils;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

/**
 * descripiton: 客戶端
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 16:40
 * @modifier:
 * @since:
 */
public class NettyClient {

    private String ip;

    private int port;

    private boolean stop = false;

    public NettyClient(String ip, int port) {
        this.ip = ip;
        this.port = port;
    }

    public void run() throws IOException {
        //設置一個多線程循環器
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        //啟動附註類
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup);
        //指定所使用的NIO傳輸channel
        bootstrap.channel(NioSocketChannel.class);
        //指定客戶端初始化處理
        bootstrap.handler(new ClientIniterHandler());
        try {
            //連接服務
            Channel channel = bootstrap.connect(ip, port).sync().channel();
            while (true) {
                //向服務端發送內容
                BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));
                String content = reader.readLine();
                if (StringUtils.isNotEmpty(content)) {
                    if (StringUtils.equalsIgnoreCase(content, "q")) {
                        System.exit(1);
                    }
                    channel.writeAndFlush(content);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            System.exit(1);
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        new NettyClient("127.0.0.1", 8899).run();
    }
}

客戶端邏輯處理：

package com.kinson.netty.client;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

/**
 * descripiton: 客戶端邏輯處理
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 16:50
 * @modifier:
 * @since:
 */
public class ClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
        //打印服務端的發送數據
        System.out.println(s);
    }
}

客戶端處理器註冊：

package com.kinson.netty.client;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

/**
 * descripiton: 客戶端處理初始化
 *
 * @author: www.iknowba.cn
 * @date: 2018/3/23
 * @time: 16:55
 * @modifier:
 * @since:
 */
public class ClientIniterHandler extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        //註冊管道
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast("decoder", new StringDecoder());
        pipeline.addLast("encoder", new StringEncoder());
        pipeline.addLast("http", new HttpClientCodec());
        pipeline.addLast("chat", new ClientHandler());
    }
}

測試時先啟動服務端，再啟動客戶端。。。

Netty實現客戶端和服務端通信簡單例子

相關推薦

Java網絡編程客戶端和服務器通信

nts cat out log auto println args etag sta 在java網絡編程中,客戶端和服務器的通信例子: 先來服務器監聽的代碼 package com.server; import java.io.IOException; import j

Netty實現客戶端和服務端通信簡單例子

啟動服務 ali tty 過程 等等 服務器初始化 讀寫操作 extends ask Netty是建立在NIO基礎之上，Netty在NIO之上又提供了更高層次的抽象。 在Netty裏面，Accept連接可以使用單獨的線程池去處理，讀寫操作又是另外的線程池來處理。 Accep

linux網絡編程之用socket實現簡單客戶端和服務端的通信（基於UDP）

服務端 msg ets lin fgets err n) stderr tcp 單客戶端和服務端的通信（基於UDP） 代碼 服務端代碼socket3.c #include<sys/types.h> #include<sys/socket.h>

Netty實現客戶端和服務端的通訊

        Netty是Apache團隊的又一個優秀的Socket框架，它和mina是一個團隊開發的，所以很多思想是相同的，接下來，我們就來實現客戶端和服務端的雙向通訊。        首先，我們定義訊息型別：/** * 訊息型別 * @author 李熠 * *

基於thrift的java和python分別作為客戶端和服務端的調用實現

Coding except arr pes com ssa utf-8 encoding 中文亂碼 前面已經實現了純java的thrift的實現。 現在實現實現一下python作為客戶端和服務端的thrift的調用 1.python作為客戶端，java作為服務端 java服

Python學習筆記1：簡單實現ssh客戶端和服務端

bsp dev bre 客戶端 break 基於 bin listen 客戶 實現基於python 3.6。 server端： 1 __author__ = "PyDev2018" 2 3 import socket,os 4 server = socket.s

Netty學習（3）: 客戶端和服務端的例子

服務端： package com.server; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.Channel; import io.netty.channel.ChannelFuture; import

python實現一個簡單的thirft客戶端和服務端

建立thrift檔案 service Hello { string get() } 使用thrift 建立服務需要的元件 thrift --gen py hello.thrif

linux網路程式設計之用socket實現簡單客戶端和服務端的通訊（基於TCP）

一、介紹基於TCP協議通過socket實現網路程式設計常用API 1、讀者如果不是很熟悉，可以先看我之前寫的幾篇部落格，有socket，地址結構的理解，更加方便讀者理解 地址分別是： 2、socket（TCP）程式設計API簡介 1）、socket int s

linux網路程式設計之用socket實現簡單客戶端和服務端的通訊（基於UDP）

1、sendto和recvfrom函式介紹 sendto（經socket傳送資料） 相關函式 send , sendmsg,recv , recvfrom , socket 表頭檔案 #include < sys/types.h >#includ

MQTTnet 實現MQTT 客戶端和服務端

服務端： class Program { private static MqttServer mqttServer = null; static void Main(string[] args) { Mqtt

利用Netty進行客戶端和服務端互動的入門示例

Netty是基於事件驅動的、非同步的網路應用程式框架和工具，用以快速開發高效能、高可靠性的網路伺服器和客戶端程式。是一種NIO框架，下面從最簡單的Hello World示例開始，演示服務端和客戶端互動進行Netty互動。 1.首先,建立服務端 public class Dis

客戶端和服務端在同一臺機器上，wireshark如何實現抓包

在實際專案中，可能會遇到客戶端和服務端在同一臺機器，要抓包對兩邊的通訊等進行分析，預設wireshark是不能抓到客戶端和服務端在同一臺機器的資料包的。 注：10.16.38.103是本地I

boost的asio實現簡單的客戶端和服務端

客戶端： //客戶端 #include "stdafx.h" #include<boost/asio/io_service.hpp> #include<boost/asio/ip/tcp.hpp> #include<

JAVA實現XML-RPC客戶端和服務端

客戶端程式碼： package test_xmlrpc.test; import java.net.URL; import java.util.ArrayList; import java.uti

socket網絡編程登錄實現及多客戶端和服務端的數據交互

取數據 sin 一個用戶 run set inf exceptio 反序列化 ram 一.TCP/IP 客戶端 package com.demo.entity; import java.io.Serializable; public class User

客戶端和服務端如何使用Token和Session

cnblogs blank style ssi exception font 統一 判斷 用戶 一、我們先解釋一下他的含義： 1、Token的引入：Token是在客戶端頻繁向服務端請求數據，服務端頻繁的去數據庫查詢用戶名和密碼並進行對比，判斷用戶名和密碼正確與否，並作

netty4----netty5的客戶端和服務端

服務端 處理 sock 一個 servers 線程不安全 inbound nio owa 服務端： package com.server; import io.netty.bootstrap.ServerBootstrap; import io.netty.channe

客戶端和服務端根路徑“/”的區別

name 根路徑 ber AC jsp 後端 ffffff http 最簡 文章來源 ： https://blog.csdn.net/zwt520123/article/details/76794446 JSP 變為 HTML 的問題 由上面可以看到，整個 Web

Linux下面安裝ftp客戶端和服務端vsftp

在Linux下面使用ftp工具，必須有客戶端和服務端。 1、使用環境客戶端為Redhat Linux6.4系統，需要到iso檔案中找到對應的Packages包，版本一定要和os版本保持一致，Linux6.4系統的ftp版本號為ftp-0.17-53.el6.x86_64.rpm ，上傳到伺服