备案控制台
探索云世界
开发者社区 开发与运维 文章 正文

Flex与Netty实现的WEB版IM(即时聊天)

2014-11-27 1572
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： Apache Flex是基于MXML和ActionScript的Flash程序设计框架，可以快速开发RIA（富Internet应用）程序，Netty是JAVA实现的高性能的网络通信框架，可以快速构建网络应用的服务端。
Apache Flex是基于MXML和ActionScript的Flash程序设计框架，可以快速开发RIA（富Internet应用）程序，Netty是JAVA实现的高性能的网络通信框架，可以快速构建网络应用的服务端。即时通讯现在已经非常普遍了，本文以简单的WEB版聊天为例浅谈IM系统的设计与原理，最后再探讨下大规模集群下的服务端的瓶颈与解决思路。

1、可用的方案选型
（1）socket
    基于浏览器的Flash插件的Socket/XMLSocket，可以采用Flex实现
（2）websocket
    HTML5的websocket是非常新的技术，但需要浏览器的支持，面临的问题是老版本的浏览器不可用。
（3）HTTP
    基于AJAX的长轮询和基于HTTP流，开源的有Comet架构的Pushlet
（4）JavaFX与Java Applet

本文采用的是方案1，用Flex编写客户端应用，目前我还没有研究如何将Flash与JS的混合，而是完全采用的Flex技术。
关于Flex的学习，我推荐到Apache上去找，也可以通过阅读《Flex实战》学习，我目前还对Flex了解很少，只能进行简单的编程，编写Flex程序前我只看了两天相关的资料，而且只用到了Socket(Flex中的一个知识点)，主要还是ActionScript的学习，ActionScript与JavaScript很类似，能看懂JS的应该也能看懂AS，这里不过多描述。

2、通讯协议设计
在即时通讯中使用的协议很多，熟知的有Jabbe/XMPP/SIP等，还有很多私有的自定义协议。协议是双方协定的数据交互规则，具体采用什么样的协议要根据场景和需求，我为了实现通讯功能，简单地设计了一个规则，具体如下：

我的使用场景包括用户的登录、注销、获取通讯录和发送消息，对于应用场景复杂的应该要考虑更多的情况，根据我在工作中的经验，通讯协议不是一下就设计好的，而是在开发过程进行不断修改与完善，可以说没有协议的设计只能遵循具体的原则，没有最终版。
我在工作中原本是基于XMPP开发的，由于做的是移动互联的应用，受限于移动网络的网速，而XMPP的协议过于庞大，对用户的流量需求太高，为此我开始寻找XMPP的替代品，我最开始选的方案是Google的ProtoBuf，类似的还有Apache的Thrift，这两者都是二进制级别的编码，虽然两者的压缩程度和性能都非常好，但在通讯协议方面不太适合，因为编码后没有可读性，出了问题不好定位。后来我在工作选了JSON来设计，相比XML来说拓展性与性能都要好很多。

3、客户端程序设计
客户端采用的是Flex，使用的集成开发工具是IntelliJ IDEA，基于Apache Flex SDK
根据通讯协议的设计进行客户端的开发，具体代码如下： 
<?xml version="1.0"?>
<s:Application xmlns:fx="http://ns.adobe.com/mxml/2009" xmlns:s="library://ns.adobe.com/flex/spark">
    <fx:Script><![CDATA[
        import mx.controls.Alert;

        var socket:Socket = null;

        // 初始化连接
        public function conn():void
        {
            socket = new Socket();
            // 增加事件监听处理
            socket.addEventListener(Event.CLOSE,closeHandler);
            socket.addEventListener(Event.CONNECT,connectHandler);
            socket.addEventListener(ProgressEvent.SOCKET_DATA, socketDataHandler);
            // 建立连接
            socket.connect('127.0.0.1',8888);
        }

        // 监听关闭事件
        private function closeHandler(event:Event):void
        {
            trace("closeHandler: " + event);
            Alert.show('closed!');
        }

        // 监听连接成功事件
        private function connectHandler(event:Event):void
        {
            trace("connectHandler: " + event);
            // 注册用户
            setName();
            Alert.show('connected!');
        }

        // 处理接收消息
        private function socketDataHandler(event:ProgressEvent):void
        {
            var str:String = socket.readUTFBytes(socket.bytesAvailable);
            trace("receive data : " + str);
            // 沙箱处理
            if(str.indexOf("<?xml version=\"1.0\"?>") == 0){
                //Alert.show(str);
            } else if (str.indexOf("ROSTER:") == 0){
                this.roster.text = str; // 处理通讯录
            } else {
                this.content.appendText(str +"\n"); // 普通消息
            }
        }

        // 点击按钮发送消息，内容为输入框中的文本
        public function send():void
        {
            var message:String = this.messageField.text;
            trace("client send : " + message);
            socket.writeUTFBytes("MSG:" + this.receive.text + "#" + this.setname.text + ":" + message);
            socket.flush();
            // 设置对话框展示效果
            this.content.appendText(this.setname.text + ":" + this.messageField.text +"\n");
            this.messageField.text = '';

        }

        // 发送字符串函数，用户注册时使用
        private function sendMsg(str:String):void
        {
            trace("client send : " + str);
            socket.writeUTFBytes(str);
            socket.flush();
        }

        // 点击关闭
        public function close():void
        {
            trace("close the connect");
            var nickname:String = this.setname.text;
            // 根据注册的用户注销用户
            sendMsg("QUIT:" + nickname);
            // 关闭连接
            socket.close();
        }

        // 设置用户名，用于注册
        public function setName():void{
            var nickname:String = this.setname.text;
            sendMsg("AUTH:" + nickname);
        }

        ]]></fx:Script>

    <s:Label text="用户名:" x="10" y="10" />
    <s:TextInput x="50" y="0" width="100" height="31" id="setname"/>
    <s:Button click="conn()" label="连接" x="160" y="0" width="60" height="31"/>

    <s:Label text="接收者:" x="10" y="50"/>
    <s:TextInput x="50" y="40" width="100" height="31" id="receive"/>
    <s:TextInput x="160" y="40" width="200" height="31" id="messageField"/>
    <s:Button click="send()" label="发送" x="370" y="40" width="60" height="31"/>
    <s:Button click="close()" label="关闭" x="440" y="40" width="60" height="31"/>

    <s:Label text="消息:" x="10" y="100"/>
    <s:Label text="通讯录:" x="320" y="100"/>
    <s:TextArea x="10" y="130" width="300" height="100" id="content"/>
    <s:TextArea x="320" y="130" width="200" height="100" id="roster"/>

</s:Application>

4、服务端设计
服务端采用的是Netty，设计方案如下：

根据方案开发的主要代码如下： 
package org.jcluster.im.server;


import org.jcluster.im.component.ComponentManager;
import org.jcluster.im.component.InterpreterComponent;
import org.jcluster.im.handler.ConnectionHandler;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
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.NioServerSocketChannel;

public class StartServer {

	private int port;

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

	public void run() throws Exception {
		// Acceptor:threads default is availableProcessors * 2
		EventLoopGroup bossGroup = new NioEventLoopGroup(2);
		// Handler
		EventLoopGroup workerGroup = new NioEventLoopGroup(4);
		try {
			ServerBootstrap server = new ServerBootstrap();
			ChannelHandler handler = new ChannelInitializer<SocketChannel>(){
				@Override
				protected void initChannel(SocketChannel ch) throws Exception {
					ch.pipeline().addLast(new ConnectionHandler());
				}
			};
			server.group(bossGroup, workerGroup)
				  .channel(NioServerSocketChannel.class)
				  .childHandler(handler)
				  .option(ChannelOption.SO_BACKLOG, 128)
				  .childOption(ChannelOption.SO_KEEPALIVE, true);
			// Start the client
			ChannelFuture future = server.bind(port).sync();
			
			InterpreterComponent component = new InterpreterComponent();
			ComponentManager.getInstance().addComponent("test", component);
			System.out.println("IM Server start");
			
			// Wait until the connection is closed
			future.channel().closeFuture().sync();
		} finally {
			workerGroup.shutdownGracefully();
			bossGroup.shutdownGracefully();
		}
	}
	
	public static void main(String[] args) throws Exception {
		new StartServer(8888).run();
		
	}
}


package org.jcluster.im.handler;

import java.net.SocketAddress;
import java.nio.charset.Charset;
import java.util.Iterator;

import org.jcluster.im.session.LocalChannelManger;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

public class ConnectionHandler extends ChannelInboundHandlerAdapter {

	@Override
	public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
		//SocketAddress address = ctx.channel().remoteAddress();
		//LocalChannelManger.getInstance().addContext(address.toString(), ctx);
		super.channelRegistered(ctx);
	}

	@Override
	public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
		LocalChannelManger.getInstance().removeContext(ctx);
		syncRoster();
		SocketAddress address = ctx.channel().remoteAddress();
		System.out.println(address.toString() + "channelUnregistered");
		int count = LocalChannelManger.getInstance().staticClients();
		System.out.println("current clients : " + count);
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		System.out.println("channelActive");
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg)
			throws Exception {
		ByteBuf in = (ByteBuf) msg;
		String message = in.toString(Charset.forName("UTF-8"));
		// Flash沙箱处理
		String xml = "<?xml version=\"1.0\"?><cross-domain-policy><site-control permitted-cross-domain-policies=\"all\"/><allow-access-from domain=\"*\" to-ports=\"*\"/></cross-domain-policy>\0";
		if(message.trim().equals("<policy-file-request/>")){
			ctx.writeAndFlush(Unpooled.copiedBuffer(xml,CharsetUtil.UTF_8));
		}
		if(message.startsWith("AUTH:")){
			String name = (message.split(":"))[1];
			LocalChannelManger.getInstance().addContext(name, ctx);
			int count = LocalChannelManger.getInstance().staticClients();
			System.out.println("current clients : " + count);
			syncRoster();
		} else if (message.startsWith("MSG:")){
			String content = message.substring(4);
			String[] temp = content.split("#");
			String to = temp[0];
			String body = "";
			for(int i=1;i<temp.length;i++){
				if(i > 1){
					body += "#";
				}
				body += temp[i];
			}
			if(LocalChannelManger.getInstance().isAvailable(to)){
				LocalChannelManger.getInstance().getContext(to).writeAndFlush(Unpooled.copiedBuffer(body,CharsetUtil.UTF_8));
			}
		} else if (message.startsWith("QUIT:")){
			String name = (message.split(":"))[1];
			LocalChannelManger.getInstance().removeContext(name);
			int count = LocalChannelManger.getInstance().staticClients();
			System.out.println("current clients : " + count);
			syncRoster();
		}
		System.out.println(message);

	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		//ctx.close();
		//System.out.println("server closed!");
	}
	
	// update all clients roster
	private void syncRoster(){
		String respone = "ROSTER:";
		for(String s : LocalChannelManger.getInstance().getAll()){
			respone += s + ",";
		}
		Iterator<ChannelHandlerContext> it = LocalChannelManger.getInstance().getAllClient().iterator();
		while(it.hasNext()){
			it.next().writeAndFlush(Unpooled.copiedBuffer(respone,CharsetUtil.UTF_8));
		}
	}

}

用户的会话管理设计：



package org.jcluster.im.session;

import io.netty.channel.ChannelHandlerContext;

import java.util.Collection;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

public class LocalChannelManger {
	// 存储用户名与连接上下文对象的映射
	final private Map<String, ChannelHandlerContext> sessions = new ConcurrentHashMap<String, ChannelHandlerContext>();
	// 存储连接上下文与用户名的映射
	final private Map<String, String> relations = new ConcurrentHashMap<String, String>();
	
	private static LocalChannelManger instance = new LocalChannelManger();
	
	public static LocalChannelManger getInstance(){
		return instance;
	}
	
	// 增加用户与连接的上下文映射
	public void addContext(String name, ChannelHandlerContext ctx){
		synchronized (sessions) {
			sessions.put(name, ctx);
			relations.put(ctx.toString(), name);
		}
	}
	
	// 获取指定用户的连接上下文
	public ChannelHandlerContext getContext(String name){
		return sessions.get(name);
	}
	
	// 根据用户名删除session
	public void removeContext(String name){
		sessions.remove(name);
	}
	
	// 判断指定的用户名当前是否在线
	public boolean isAvailable(String name){
		return sessions.containsKey(name) && (sessions.get(name) != null);
	}
	
	// 获取所有的用户名
	public synchronized Set<String> getAll(){
		return sessions.keySet();
	}
	
	// 获取所有连接的上下文对象
	public synchronized Collection<ChannelHandlerContext> getAllClient(){
		return sessions.values();
	}
	
	// 根据上下文删除用户session
	public void removeContext(ChannelHandlerContext ctx){
		String name = relations.get(ctx.toString());
		if(name != null){
			sessions.remove(name);
			relations.remove(ctx.toString());
		}
	}
	
	// 统计当前在线人数
	public int staticClients(){
		return relations.size();
	}
	
}

5、系统可拓展性

当用户量大时，服务端布署就多了，实际生产环境下是有server to server的消息转发，即登录在一台的用户发消息给登录在另一台的用户，这时候系统的瓶颈就出现了，我模仿网络中TCP报文的路由思想，写了一个路由策略，具体代码如下：

package org.jcluster.im;

import java.util.Map;
import java.util.Map.Entry;

/**
 * 消息路由器
 * @author zhaowen
 *
 */
public class RouteServer {
	// key:IP, value:IP and UDP Port
	private Map<String,String> routeTable;
	// userId/device,chanelId
	private Map<String,String> localSession;
	// off-line message push
	private OfflinePush offlinePush;
	// local session route
	private LocalRouter localRouter;
	
	public void route(Package pack){
		if(localSession.containsKey(pack.getTo().toString())){
			// localRoute
			localRouter.route(pack);
		} else {
			// route track count
			int count = 0;
			for(Entry<String,String> node : routeTable.entrySet()){
				if(!pack.getTrack().containsKey(node.getKey())){
					// localHost add to the track
					pack.getTrack().put("localNode", "IP");
					// route to the next RouteServer though UDP or MQ
					UDPUtil.send(pack, node.getValue());
				} else {
					// the route track count+1
					count++;
				}
			}
			if(count >= routeTable.size()){
				// off-line message push
				offlinePush.push(pack.getMsg());
			}
		}
	}
	
	// 观察者模式,消息监听器
	public class Task implements RouteListener{

		@Override
		public void receive(Package pack) {
			route(pack);
		}
	}
	
	public Task newTask(){
		return new Task();
	}
	
}

6、测试

服务器日志：

IM Server start
channelActive
current clients : 1
AUTH:hello
channelActive
current clients : 2
AUTH:jj
MSG:hello#jj:dfasd#asddsf#df$%^#$^$#^
MSG:jj#hello:sdfas#@$@#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:dfasd#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:dsfad#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&
MSG:jj#hello:#SDGSgQ#$!@#$!^$%&sdfsafs
MSG:jj#hello:asdf#SDGSgQ#$!@#$!^dfasfsdf$%&
/127.0.0.1:60385channelUnregistered
current clients : 1
/127.0.0.1:60386channelUnregistered
current clients : 0


文章标签：
客服工作台
Java
Web App开发
数据格式
XML
Apache
关键词：
web im
客服工作台web
Netty im
客服工作台netty
Netty web
凌峯
目录
相关文章
jsjsjjs
|
10月前
|
存储 缓存 NoSQL
跟着源码学IM(十一)：一套基于Netty的分布式高可用IM详细设计与实现(有源码)
本文将要分享的是如何从零实现一套基于Netty框架的分布式高可用IM系统，它将支持长连接网关管理、单聊、群聊、聊天记录查询、离线消息存储、消息推送、心跳、分布式唯一ID、红包、消息同步等功能，并且还支持集群部署。
jsjsjjs
13133 1
java知路
|
7月前
|
Java Maven Spring
使用netty实现nio web服务器
使用netty实现nio web服务器
java知路
59 0
jsjsjjs
|
4月前
|
前端开发 安全 数据安全/隐私保护
Web网页端IM产品RainbowChat-Web的v6.0版已发布
RainbowChat-Web是一套Web网页端IM系统，是RainbowChat的姊妹系统（RainbowChat是一套基于开源IM聊天框架 MobileIMSDK (Github地址) 的产品级移动端IM系统）。
jsjsjjs
35 0
不焦躁的程序员
|
4月前
|
JSON 算法 Dubbo
Netty入门实践-模拟IM聊天
本文以入门实践为主，通过原理+代码的方式，实现一个简易IM聊天功能。分为2个部分：Netty的核心概念、IM聊天简易实现。
不焦躁的程序员
108 0
jsjsjjs
|
4月前
|
NoSQL Java Redis
跟着源码学IM(十二)：基于Netty打造一款高性能的IM即时通讯程序
关于Netty网络框架的内容，前面已经讲了两个章节，但总归来说难以真正掌握，毕竟只是对其中一个个组件进行讲解，很难让诸位将其串起来形成一条线，所以本章中则会结合实战案例，对Netty进行更深层次的学习与掌握，实战案例也并不难，一个非常朴素的IM聊天程序。 原本打算做个多人斗地主练习程序，但那需要织入过多的业务逻辑，因此一方面会带来不必要的理解难度，让案例更为复杂化，另一方面代码量也会偏多，所以最终依旧选择实现基本的IM聊天程序，既简单，又能加深对Netty的理解。
jsjsjjs
85 1
jsjsjjs
|
10月前
|
SQL 前端开发 安全
Web网页端IM产品RainbowChat-Web的v5.0版已发布
Web网页端IM产品RainbowChat-Web的v5.0版已发布
jsjsjjs
101 0
-编程工程师-
|
12月前
|
消息中间件 前端开发 JavaScript
太顶了，使用 Netty 实现了一个 IM 即时通讯系统
太顶了，使用 Netty 实现了一个 IM 即时通讯系统
-编程工程师-
662 0
jsjsjjs
|
移动开发 安全 网络协议
手把手教你为基于Netty的IM生成自签名SSL/TLS证书
本文要分享的是如何使用OpenSSL生成在基于Netty的IM中真正可用的SSL/TLS证书，内容包括：证书的创建、创建过程中的注意点，以及在Server端、Android端、iOS端、Java桌面端、H5端使用证书的代码范例。
jsjsjjs
343 0
手把手教你为基于Netty的IM生成自签名SSL/TLS证书
i校长
|
测试技术 iOS开发
Flutter Web网站之最简方式实现暗黑主题无缝切换
Flutter Web网站之最简方式实现暗黑主题无缝切换
i校长
268 0
Flutter Web网站之最简方式实现暗黑主题无缝切换
章为忠学架构
|
消息中间件 网络协议 前端开发
SpringBoot轻松整合WebSocket，实现Web在线聊天室
前面为大家讲述了 Spring Boot的整合Redis、RabbitMQ、Elasticsearch等各种框架组件；随着移动互联网的发展，服务端消息数据推送已经是一个非常重要、非常普遍的基础功能。今天就和大家聊聊在SpringBoot轻松整合WebSocket，实现Web在线聊天室，希望能对大家有所帮助。
章为忠学架构
798 1
SpringBoot轻松整合WebSocket，实现Web在线聊天室

热门文章

最新文章

  • 1
    除了 Mattermost，这些开源 IM 应用也值得一试
  • 2
    工单系统大揭秘！选择工单系统需注意的关键因素！
  • 3
    人工智能，应该如何测试？（七）大模型客服系统测试
  • 4
    Netty网络框架（一）
  • 5
    阿里云安全产品，Web应用防火墙与云防火墙产品各自作用简介
  • 6
    构建高效响应式Web界面：现代前端框架的比较
  • 7
    Netty的基础入门（上）
  • 8
    Serverless 应用引擎常见问题之做的web服务计费如何解决
  • 9
    机器学习PAI常见问题之web ui 项目启动后页面打不开如何解决
  • 10
    netty实现tomcat（简易版）
  • 1
    实验目的1.编译安装httpd2.优化路径3.并将鲜花网站上传到web服务器为网页目录4.在客户机访问网站http://www.bdqn.com
    163
  • 2
    构建高效Python Web应用：异步编程与Tornado框架解析
    28
  • 3
    如何在 Python 中进行 Web 应用程序的安全性管理，例如防止 SQL 注入？
    15
  • 4
    请描述在 Python WEB 开发中常用的测试方法。
    16
  • 5
    在 Python 中，如何处理 Web 请求和响应？
    26
  • 6
    Django是一个用Python编写的开源Web应用框架
    12
  • 7
    什么是JWT(Json-Web-Token)？JWT的用途和优势是什么？讲解+实战，一篇文章学会JWT怎么用！
    36
  • 8
    构建高性能Web应用：前端性能优化的关键策略与实践
    41
  • 9
    构建响应式Web应用：现代前端框架的比较与实践
    20
  • 10
    视觉智能平台常见问题之web端编辑器实现如何解决
    36

    • 相关课程

    更多
  • Python Web开发基础
  • Java Web开发系列课程 - Spring框架入门
  • 企业Web常用架构LAMP-LNMP实战
  • 高校精品课-西安交通大学-Web 编程技术
  • Java Web项目实战 - 图书商城
  • Java面试疑难点解析 - Java Web开发

    • 相关电子书

    更多
  • Phoenix在客服系统的实践
  • # Apache Spark系列技术直播# 第八讲 【 微软Azure平台利用Intel Analytics Zoo构建AI客服支持实践 】
  • 客服领域的数据智能实践

    • 相关实验场景

    更多
  • 使用SLB+2ECS+NAS，部署电商web网站的高可用架构
  • 1分钟SAE部署Web在线游戏
  • 云上Web及FTP
  • WEB网页编程实战
  • 通过部署流行Web框架掌握Serverless技术
  • 手动部署Java Web环境（Alibaba Cloud Linux 2）
    • 下一篇
    阿里云oss简介和使用流程