Netty是一个基于NIO,异步的,事件驱动的网络通信框架。由于使用Java提供 的NIO包中的API开发网络服务器代码量大,复杂,难保证稳定性。netty这类的网络框架应运而生。通过使用netty框架可以快速开发网络通信服务端,客户端。

  本文主要通过一个简单的聊天程序来熟悉初步使用Nettyty进行简单服务端与客户端的开发。本聊天系统主要功能有点对点聊天及服务端推送消息。

程序结构:

   Server端: IMServer 服务器启动类   ServerHandler 服务端核心类 负责客户端认证及消息转发

   Client端:  IMClient 客户端启动类    ClientHandler 客户端核心类,负责客户端消息的发送及接收

  Coder:MsgPackDecode和MsgPackEncode,负责消息的解码及编码实现,消息的编解码基于第三方库msgpack 

代码分析:

   代码结构如下

    

ApplicationContext

功能比较简单,主要用来保存登录用户信息,以Map来存储,其中key为用户ID,value为客户端对应的ChannelHandlerContext对象。

import io.netty.channel.ChannelHandlerContext;
import java.util.HashMap;
import java.util.Map; /**
* Created by Andy on 2016/10/8.
*/
public class ApplicationContext {
public static Map<Integer,ChannelHandlerContext> onlineUsers = new HashMap<Integer,ChannelHandlerContext>();
public static void add(Integer uid,ChannelHandlerContext ctx){
onlineUsers.put(uid,ctx);
} public static void remove(Integer uid){
onlineUsers.remove(uid);
} public static ChannelHandlerContext getContext(Integer uid){
return onlineUsers.get(uid);
}
}

IMServerConfig接口

该接口主要用来存储服务端启动的配置信息,可改为配置文件实现

import com.wavemelody.nettyim.struts.MessageType;

/**
* Created by Andy on 2016/10/8.
*/
public interface IMServerConfig {
/**客户端配置*/
int CLIENT_VERSION = 1; //版本号
/**服务端配置*/
String SERVER_HOST = "127.0.0.1"; //服务器IP
int SERVER_PORT = 9090; //服务器端口
/**消息相关*/
int SERVER_ID = 0; //表示服务器消息
byte APP_IM = 1; //即时通信应用ID为1
MessageType TYPE_MSG_CONNECT = MessageType.TYPE_AUTH; //连接后第一次消息确认建立连接和发送认证信息
MessageType TYPE_MSG_TEXT = MessageType.TYPE_TEXT; //文本消息
String MSG_DEFAULT = ""; //空消息
}

  

ServerHandler

服务端主要的消息处理Handler,负责客户端认证之后,对客户端信息的保存及客户端点对点消息的转发以及程序异常时对资源的关闭等业务。

import com.wavemelody.nettyim.server.core.ApplicationContext;
import com.wavemelody.nettyim.struts.IMMessage;
import com.wavemelody.nettyim.struts.MessageType;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter; /**
* Created by Andy on 2016/10/8.
*/
public class ServerHandler extends ChannelInboundHandlerAdapter{
private ChannelHandlerContext ctx; @Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
System.out.println("服务端Handler创建。。。");
super.handlerAdded(ctx);
} @Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("channelInactive");
super.channelInactive(ctx);
} @Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
this.ctx = ctx;
super.channelActive(ctx);
System.out.println("有客户端连接:" + ctx.channel().remoteAddress().toString());
} @Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
IMMessage message = (IMMessage)msg;
if(message.getMsgType() == MessageType.TYPE_AUTH.value()){ //认证消息
System.out.println("认证消息:" + msg);
ApplicationContext.add(message.getUid(),ctx);
}else if(message.getMsgType() == MessageType.TYPE_TEXT.value()){ //CHAT消息
ChannelHandlerContext c = ApplicationContext.getContext(message.getReceiveId());
if(c==null){ //接收方不在线,反馈给客户端
message.setMsg("对方不在线!");
ctx.writeAndFlush(message);
}else{ //将消转发给接收方
System.out.println("转发消息:" + msg);
c.writeAndFlush(message);
}
} } @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("与客户端断开连接:"+cause.getMessage());
cause.printStackTrace();
ctx.close();
} }

  

IMServer

服务端的启动类,关于服务端,有以下几点需要说明

  1. runServerCMD()方法用来启动控制台,启动后,可以对用户输入的内容进行消息推送。
  2. MsgPackEncode和MsgPackDecode用于消息的编解码。使用的是MessagePack(API使用简单,编码后字节流特小,编解码速度较快,同时几乎支持所有主流编程语言,详情见官网:http://msgpack.org/)。这样我们可以随意编写实体用于发送消息,相关代码后边给出。
  3. LengthFieldBasedFrameDecoder和LengthFieldPrepender:因为TCP底层传输数据时是不了解上层业务的,所以传输消息的时候很容易造成粘包/半包的情况(一条完整的消息被拆开或者完整或者不完整的多条消息被合并到一起发送、接收),这两个工具就是Netty提供的消息编码工具,2表示消息长度(不是正真的长度为2,是2个字节)。
import com.wavemelody.nettyim.codec.MsgPackDecode;
import com.wavemelody.nettyim.codec.MsgPackEncode;
import com.wavemelody.nettyim.server.config.IMServerConfig;
import com.wavemelody.nettyim.server.core.ApplicationContext;
import com.wavemelody.nettyim.server.handler.ServerHandler;
import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.io.IOException;
import java.util.Map;
import java.util.Scanner; /**
* Created by Andy on 2016/10/8.
*/
public class IMServer implements Runnable,IMServerConfig{
public static void main(String[] args) throws IOException{
new IMServer().start();
}
public void start() throws IOException{
new Thread(this).start();
runServerCMD();
} private IMMessage getMessage(){
int toID = -1;
IMMessage message = new IMMessage(
APP_IM,
CLIENT_VERSION,
SERVER_ID,
TYPE_MSG_TEXT.value(),
toID,
MSG_DEFAULT);
return message;
} private void runServerCMD()throws IOException{ Scanner scanner = new Scanner(System.in);
IMMessage message = null;
do{
message = getMessage();
message.setMsg(scanner.nextLine());
}while(sendMsg(message));
} private boolean sendMsg(IMMessage msg){
// 当用户输入quit表示退出,不在进行推送
boolean result = msg.getMsg().equals("quit") ? false:true;
if(result){
int receiveID = msg.getReceiveId();
String content = msg.getMsg();
if(content.startsWith("#") && content.indexOf(":") != -1){
try {
/**
* 用户输入指定的推送客户端
* 输入文本格式为: "#8888:发送内容"
* “#”和“:”之间内容为用户ID,“:”之后为推送消息内容
*/
receiveID = Integer.valueOf(content.substring(1,content.indexOf(":")));
msg.setReceiveId(receiveID);
msg.setMsg(content.substring(content.indexOf(":")));
} catch (NumberFormatException e) {
//解析失败则,默认发送所有
e.printStackTrace();
} } /**
* 默认推送所有用户(默认receiveID为-1)
* */
if(receiveID == -1){
System.out.println("推送消息给所有在线用户:" + msg);
for(Map.Entry<Integer,ChannelHandlerContext> entry: ApplicationContext.onlineUsers.entrySet()){
ChannelHandlerContext c = entry.getValue();
c.writeAndFlush(msg);
}
}else{
ChannelHandlerContext ctx = ApplicationContext.getContext(receiveID);
if(ctx!=null){
System.out.println("推送消息:" + msg);
ctx.writeAndFlush(msg);
} } }
return result;
} @Override
public void run() {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup(); try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG,1024)
.childOption(ChannelOption.SO_KEEPALIVE, true)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(65536, 0, 2, 0, 2));
ch.pipeline().addLast("msgpack decoder",new MsgPackDecode());
ch.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
ch.pipeline().addLast("msgpack encoder",new MsgPackEncode());
ch.pipeline().addLast(new ServerHandler());
}
});
ChannelFuture f = b.bind(SERVER_PORT).sync();
f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}

ClientHandler

客户端Handler

import com.wavemelody.nettyim.client.config.IMClientConfig;
import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.io.IOException; /**
* Created by Andy on 2016/10/8.
*/
public class ClientHandler extends ChannelInboundHandlerAdapter implements IMClientConfig{
private ChannelHandlerContext ctx; @Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("用户["+ UID + "]成功连接服务器");
this.ctx = ctx; //通道建立时发送认证消息给服务器
IMMessage message = new IMMessage(
APP_IM,
CLIENT_VERSION,
UID,
TYPE_MSG_AUTH.value(),
DEFAULT_RECEIVE_ID,
MSG_DEFAULT);
sendMsg(message);
} public boolean sendMsg(IMMessage msg) throws IOException {
boolean result = msg.getMsg().equals("quit") ? false:true;
if(result){
if(msg.getMsgType() != MessageType.TYPE_AUTH.value()){
System.out.println("认证消息: " + "client[" + msg.getUid() + "]:" + msg.getMsg());
}
//设置接收端ID和发送消息
if(msg.getMsgType() == MessageType.TYPE_TEXT.value()){
if(msg.getMsg().contains(":")){
String[] msgs = msg.getMsg().split(":");
String receiveIdStr =msgs[0].substring(1);
msg.setReceiveId(Integer.valueOf(receiveIdStr));
msg.setMsg(msgs[1]);
}
}
ctx.writeAndFlush(msg);
}
return result;
} @Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
IMMessage m = (IMMessage)msg;
System.out.println("receive[" + m.getUid() + "]:" + m.getMsg());
} @Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("与服务器断开连接:" + cause.getMessage());
ctx.close();
}
}

IMClient

客户端启动类,同时启动控制台,将用户输入消息发送给指定的客户端

import com.wavemelody.nettyim.client.config.IMClientConfig;
import com.wavemelody.nettyim.client.handler.ClientHandler;
import com.wavemelody.nettyim.codec.MsgPackDecode;
import com.wavemelody.nettyim.codec.MsgPackEncode;
import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import java.io.IOException;
import java.util.Scanner; /**
* Created by Andy on 2016/10/8.
*/
public class IMClient implements Runnable,IMClientConfig{
private ClientHandler clientHandler = new ClientHandler();
public static void main(String[] args) throws IOException{
new IMClient().start();
} public void start() throws IOException{
new Thread(this).start();
runClientCMD();
}
public void runClientCMD() throws IOException{
IMMessage message = new IMMessage(
APP_IM,
CLIENT_VERSION,
UID,
TYPE_MSG_TEXT.value(),
DEFAULT_RECEIVE_ID,
MSG_DEFAULT);
Scanner scanner = new Scanner(System.in);
do{
message.setMsg(scanner.nextLine());
}
while (clientHandler.sendMsg(message));
} @Override
public void run() {
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(workerGroup)
.channel(NioSocketChannel.class)
.option(ChannelOption.SO_KEEPALIVE, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65536, 0, 2, 0, 2));
ch.pipeline().addLast("msgpack decoder",new MsgPackDecode());
ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(2));
ch.pipeline().addLast("msgpack encoder",new MsgPackEncode());
ch.pipeline().addLast(clientHandler);
}
});
ChannelFuture f = b.connect(SERVER_HOST, SERVER_PORT).sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
}
}
}

IMMessage类

该类即为通讯过程中的消息实体,即通讯协议,客户端进行消息发送,服务端进行消息推送时都需要将发送内容封装为IMMessage对象,才可以被识别。

import org.msgpack.annotation.Message;

/**
* Created by Andy on 2016/10/8.
*/
@Message
public class IMMessage {
//应用ID
private byte appId; //版本
private int version; //用户ID
private int uid; //消息类型 0:登录 1:文字消息
private byte msgType; //接收方
private int receiveId; //消息内容
private String msg; public IMMessage(){ } /**
* 构造方法
* @param appId 应用通道
* @param version 应用版本
* @param uid 用户ID
* @param msgType 消息类型
* @param receiveId 消息接收者
* @param msg 消息内容
*/
public IMMessage(byte appId, int version, int uid, byte msgType, int receiveId, String msg) {
this.appId = appId;
this.version = version;
this.uid = uid;
this.msgType = msgType;
this.receiveId = receiveId;
this.msg = msg;
} public byte getAppId() {
return appId;
} public void setAppId(byte appId) {
this.appId = appId;
} public int getVersion() {
return version;
} public void setVersion(int version) {
this.version = version;
} public int getUid() {
return uid;
} public void setUid(int uid) {
this.uid = uid;
} public byte getMsgType() {
return msgType;
} public void setMsgType(byte msgType) {
this.msgType = msgType;
} public int getReceiveId() {
return receiveId;
} public void setReceiveId(int receiveId) {
this.receiveId = receiveId;
} public String getMsg() {
return msg;
} public void setMsg(String msg) {
this.msg = msg;
} @Override
public String toString() {
return "IMMessage{" +
"appId=" + appId +
", version=" + version +
", uid=" + uid +
", msgType=" + msgType +
", receiveId=" + receiveId +
", msg='" + msg + '\'' +
'}';
}
}

MessageType

消息类型,通过枚举类型来约束消息中消息类型字段内容,防止出现系统不能识别的消息类型而发生异常。

/**
* Created by Andy on 2016/10/9.
*/
public enum MessageType {
TYPE_AUTH((byte)0),TYPE_LOGOUT((byte)1),TYPE_TEXT((byte)2),TYPE_EMPTY((byte)3);
private byte value;
MessageType(byte value){
this.value = value;
}
public byte value(){
return this.value;
}
}

IMClientConfig接口

主要用来定义客户端启动配置信息常量,可改为配置文件实现方式。

import com.wavemelody.nettyim.struts.MessageType;

/**
* Created by Andy on 2016/10/9.
*/
public interface IMClientConfig {
/**客户端配置*/
int CLIENT_VERSION = 1; //版本号
/**服务端配置*/
String SERVER_HOST = "127.0.0.1"; //服务器IP
int SERVER_PORT = 9090; //服务器端口
/**消息相关*/
byte APP_IM = 1; //即时通信应用ID为1 int UID = 8888;
int DEFAULT_RECEIVE_ID = 9999; MessageType TYPE_MSG_AUTH = MessageType.TYPE_AUTH; //连接后第一次消息确认建立连接和发送认证信息
MessageType TYPE_MSG_TEXT = MessageType.TYPE_TEXT; //文本消息
String MSG_DEFAULT = ""; //默认为空消息
}

MsgPackEncode

使用msgpack实现对消息的编码实现。

import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.msgpack.MessagePack; /**
* Created by Andy on 2016/10/8.
*/
public class MsgPackEncode extends MessageToByteEncoder<IMMessage> {
@Override
protected void encode(ChannelHandlerContext channelHandlerContext, IMMessage msg, ByteBuf out) throws Exception {
out.writeBytes(new MessagePack().write(msg));
}
}

MsgPackDecode

使用msgpack实现对消息的解码实现。

import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
import org.msgpack.MessagePack; import java.util.List; /**
* Created by Andy on 2016/10/8.
*/
public class MsgPackDecode extends MessageToMessageDecoder<ByteBuf>{ @Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf msg, List<Object> out) throws Exception {
final int length = msg.readableBytes();
final byte[] array = new byte[length];
msg.getBytes(msg.readerIndex(),array,0,length);
out.add(new MessagePack().read(array, IMMessage.class));
}
}

  通过以上代码基本上已经实现了一个简单的聊天程序,当然还存在很多地方需要优化。一个就是对TCP连接的优化,不是指定SO_KEEPALIVE属性,而是改为发送心跳消息来维持客户端和服务器的连接;然后就是链路中断后的重连实现,当出现中断之后由客户端等待一定时间重新发起连接操作,直至连接成功;另外一个就是重复登录验证,在客户端已经登录的情况下,要拒绝重复登录,防止客户端在异常状态下反复重连导致句柄资源被耗尽。  


    

  

Netty实现一个简单聊天系统(点对点及服务端推送)的更多相关文章

  1. java SDK服务端推送 --极光推送(JPush)

    网址:https://blog.csdn.net/duyusean/article/details/86581475 消息推送在APP应用中越来越普遍,来记录一下项目中用到的一种推送方式,对于Andr ...

  2. [译]servlet3.0与non-blocking服务端推送技术

    Non-blocking(NIO)Server Push and Servlet 3 在我的前一篇文章写道如何期待成熟的使用node.js.假定有一个框架,基于该框架,开发者只需要定义协议及相关的ha ...

  3. 一文了解服务端推送(含JS代码示例)

    常用的服务端推送技术,包括轮询.长轮询.websocket.server-sent-event(SSE) 传统的HTTP请求是由客户端发送一个request,服务端返回对应response,所以当服务 ...

  4. mqtt协议实现 java服务端推送功能(三)项目中给多个用户推送功能

    接着上一篇说,上一篇的TOPIC是写死的,然而在实际项目中要给不同用户 也就是不同的topic进行推送 所以要写活 package com.fh.controller.information.push ...

  5. 升级NGINX支持HTTP/2服务端推送

    内容概览 NGINX从1.13.9版本开始支持HTTP/2服务端推送,上周找时间升级了下NGINX,在博客上试验新的特性. 升级工作主要包括: 升级NGINX 修改NGINX配置 修改wordpres ...

  6. C# 服务端推送,十步十分钟,从注册到推送成功

    目标 展示 C# 服务端集成极光推送的步骤,多图少字,有图有真相. 使用极光推送, C# 服务端推送到 Demo App,Android 手机收到推送,整理为十个步骤,使用十分钟左右,完成从注册账号到 ...

  7. 利用WebSocket和EventSource实现服务端推送

    可能有很多的同学有用 setInterval 控制 ajax 不断向服务端请求最新数据的经历(轮询)看下面的代码: setInterval(function() { $.get('/get/data- ...

  8. 用socket写一个简单的客户端和服务端程序

    用来练手写写socket代码 客户端代码 #include <stdio.h> #include <sys/types.h> #include <sys/socket.h ...

  9. http2 技术整理 nginx 搭建 http2 wireshark 抓包分析 server push 服务端推送

    使用 nginx 搭建一个 http2 的站点,准备所需: 1,域名 .com .net 均可(国内域名需要 icp 备案) 2,云主机一个,可以自由的安装配置软件的服务器 3,https 证书 ht ...

随机推荐

  1. Android端接收和发送cookie

    流程: 首先android端使用HttpClient的方式发送HTTP请求,此时服务器创立cookie,并发送cookie给android端,android端再将cookie保存起来,在需要发送coo ...

  2. vs2010 用户控件拖到aspx页面不可用

    错误描述: 在web项目中添加一个用户控件,直接拖动用户控件ascx到aspx页面出现a标签而不是控件标签 解决办法: 把“源”切换为“设计”视图,然后拖动ascx用户控件到页面即可:

  3. 七、context command

    context command是用来新建自己的工具,可以调用OPENGL,获取鼠标操作函数,在view窗口画自己想画的东西.(我是这麽理解的,可以以后再确定一下) 下面是一个context comma ...

  4. sql server 复制表结构

    1:复制表结构及数据到新表 select * into 目的数据库名.dbo.目的表名 from 原表名 select * into my0735home.dbo.infoMianTest from ...

  5. BZOJ4974 八月月赛 Problem D 字符串大师 KMP

    欢迎访问~原文出处——博客园-zhouzhendong 去博客园看该题解 题目传送门 - BZOJ4974 - 八月月赛 Problem D 题意概括 一个串T是S的循环节,当且仅当存在正整数k,使得 ...

  6. Asp.Net前台调用后台变量

    1.Asp.Net中几种相似的标记符号: < %=...%>< %#... %>< % %>< %@ %>解释及用法 答: < %#... %&g ...

  7. Zabbix 添加主机,获取模板templateID

    添加一个Host Name=Mail CAS_1.1,IP=10.16.3.4的主机,并加入组groupID=30,连接模板templateID=10132. # -*- coding: UTF-8 ...

  8. OAuth 2.0 Salesforce &amp; Azure

    最近在学习Salesforce,浅谈一下 OAuth 2.0 在Salesforce and Azure 之间的应用. 假设有这样一个场景,在Salesforce中需要用到Azure中的一些服务,那么 ...

  9. MATLAB 制作GIF图像

    前提要求:图像集保存在某个文件夹中,且每个图像以数字形式顺序命名,如001.jpg,002.jpg等. 代码1: 这个代码生成的效果有点问题,建议采用代码2. wm={'overwrite','app ...

  10. JDK5.0 特性-线程同步装置之Semaphore

    来自:http://www.cnblogs.com/taven/archive/2011/12/17/2291474.html import java.util.ArrayList; import j ...