功能介绍
通过数据算法提高商品在消费者搜索结果页，提升商品的曝光率（利用手淘访问的缓存机制提供商品排名的接口）。
#功能用途
1、配合店铺活动将活动商品推送到潜在购买的消费者搜索结果里去，增加商品的曝光率，提高交易额。

2、配合第三方补单平台，在短词搜索结果中出现推送的商品

3、推送给意向客户，之前有咨询没付款的，使用权重推送后，能将宝贝推送到他们的搜索页。

商

品权重越高推送成功率越大，消费者也很难在首屏看到权重较低商品。有基础权重的商品推成功率较高，一旦显示推送成功，推送显示成功的概率高达90%，介意勿用该功能
搜索权重推送api

接口地址： http://xxxxxxxxxxxxxxxxxxxxx/searchpush

接口介绍: 模拟官方算法提升指定淘宝号搜索结果页里某个商品排名

巍信 a2061880542

请求方式： POST

传参方式： form-data

返回格式： JSON

接口收费： 内测用户，查询1次，计1次调用

频率限制： 10次/秒


请求头参数





参数
类型
必选
值
示例




Authorization
string
是
授权 access_token
Bearer 1798c6aadec33d1bc2f5b707f1049aefexxxx



请求参数





参数
类型
必选
描述




items.url
string
是
商品id


items.tag
string
是
关键字


items.account
string
是
淘宝账号



请求示例

注意使用form-data方式传参，key 为 items
[   
    {"account":"马云","tag":"衣柜","url":"573712069552"},
    {"account":"芭比博士","tag":"衣柜","url":"573712069552"}
]


cURL请求代码示例


curl --location --request POST 'http://api.vv-tool.com/tool/accounts/searchpush' \
--header 'Cookie: Hm_lvt_eaa57ca47dacb4ad4f5a257001a3457c=1594955822,1595049020,1595211254; Hm_lvt_b2f2f0b6e89c111ec6ae8a8ffe0438b4=1595811470,1595898219,1595983911,1596070884; Hm_lpvt_b2f2f0b6e89c111ec6ae8a8ffe0438b4=1596079373; _csrf=ef485c09be934be843408cd625cc5d60ba5e265e9a0f0e707a589ec8f6862f57a%3A2%3A%7Bi%3A0%3Bs%3A5%3A%22_csrf%22%3Bi%3A1%3Bs%3A32%3A%22vNSc2nlQlgzl7vZ8sXOel3RnVCYT3LhL%22%3B%7D; PHPSESSID=9clrl8itapsmakj80smt4rr0ts; _csrf=5f72449d88f0ff6404d5eb64f3722b8c7dde2fab1ec9f0f10c7a5808cb14eab0a%3A2%3A%7Bi%3A0%3Bs%3A5%3A%22_csrf%22%3Bi%3A1%3Bs%3A32%3A%22a1vK_3_BGOjXX6smQ5tPL1b33FB59M5G%22%3B%7D' \
--header 'Authorization: Bearer c2c20409e5f3f34437aece02c69b9a9dxxxxxxxx' \
--form 'items=[{"account":"马云","tag":"衣柜","url":"573712069552"},{"account":"芭比博士","tag":"衣柜","url":"573712069552"}]'



PHP请求代码示例


<?php

$curl = curl_init();

curl_setopt_array($curl, array(
  CURLOPT_URL => "http://api.vv-tool.com/tool/accounts/searchpush",
  CURLOPT_RETURNTRANSFER => true,
  CURLOPT_ENCODING => "",
  CURLOPT_MAXREDIRS => 10,
  CURLOPT_TIMEOUT => 0,
  CURLOPT_FOLLOWLOCATION => true,
  CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
  CURLOPT_CUSTOMREQUEST => "POST",
  CURLOPT_POSTFIELDS => array('items' => '[{"account":"马云","tag":"衣柜","url":"573712069552"},{"account":"芭比博士","tag":"衣柜","url":"573712069552"}]'),
  CURLOPT_HTTPHEADER => array(
    "Cookie: Hm_lvt_eaa57ca47dacb4ad4f5a257001a3457c=1594955822,1595049020,1595211254; Hm_lvt_b2f2f0b6e89c111ec6ae8a8ffe0438b4=1595811470,1595898219,1595983911,1596070884; Hm_lpvt_b2f2f0b6e89c111ec6ae8a8ffe0438b4=1596079373; _csrf=ef485c09be934be843408cd625cc5d60ba5e265e9a0f0e707a589ec8f6862f57a%3A2%3A%7Bi%3A0%3Bs%3A5%3A%22_csrf%22%3Bi%3A1%3Bs%3A32%3A%22vNSc2nlQlgzl7vZ8sXOel3RnVCYT3LhL%22%3B%7D; PHPSESSID=9clrl8itapsmakj80smt4rr0ts; _csrf=5f72449d88f0ff6404d5eb64f3722b8c7dde2fab1ec9f0f10c7a5808cb14eab0a%3A2%3A%7Bi%3A0%3Bs%3A5%3A%22_csrf%22%3Bi%3A1%3Bs%3A32%3A%22a1vK_3_BGOjXX6smQ5tPL1b33FB59M5G%22%3B%7D",
    "Authorization: Bearer c2c20409e5f3f34437aece02c69b9a9dxxxxxxxx"
  ),
));

$response = curl_exec($curl);

curl_close($curl);
echo $response;



java请求代码示例


OkHttpClient client = new OkHttpClient().newBuilder()
  .build();
MediaType mediaType = MediaType.parse("text/plain");
RequestBody body = new MultipartBody.Builder().setType(MultipartBody.FORM)
  .addFormDataPart("items", "[{\"account\":\"马云\",\"tag\":\"衣柜\",\"url\":\"573712069552\"},{\"account\":\"芭比博士\",\"tag\":\"衣柜\",\"url\":\"573712069552\"}]")
  .build();
Request request = new Request.Builder()
  .url("http://api.vv-tool.com/tool/accounts/searchpush")
  .method("POST", body)
  .addHeader("Cookie", "Hm_lvt_eaa57ca47dacb4ad4f5a257001a3457c=1594955822,1595049020,1595211254; Hm_lvt_b2f2f0b6e89c111ec6ae8a8ffe0438b4=1595811470,1595898219,1595983911,1596070884; Hm_lpvt_b2f2f0b6e89c111ec6ae8a8ffe0438b4=1596079373; _csrf=ef485c09be934be843408cd625cc5d60ba5e265e9a0f0e707a589ec8f6862f57a%3A2%3A%7Bi%3A0%3Bs%3A5%3A%22_csrf%22%3Bi%3A1%3Bs%3A32%3A%22vNSc2nlQlgzl7vZ8sXOel3RnVCYT3LhL%22%3B%7D; PHPSESSID=9clrl8itapsmakj80smt4rr0ts; _csrf=5f72449d88f0ff6404d5eb64f3722b8c7dde2fab1ec9f0f10c7a5808cb14eab0a%3A2%3A%7Bi%3A0%3Bs%3A5%3A%22_csrf%22%3Bi%3A1%3Bs%3A32%3A%22a1vK_3_BGOjXX6smQ5tPL1b33FB59M5G%22%3B%7D")
  .addHeader("Authorization", "Bearer 4eaa83b3d669a05ca87b61fb72537839xxxxxxx")
  .build();
Response response = client.newCall(request).execute();



响应参数





参数
类型
描述




code
int
返回码


msg
string
返回码描述



返回数据


    {
      "code": 0,
      "msg": "成功"
    }



响应代码





代码
代码描述
解决方案




0
成功



100
请求参数错误
检查您的请求参数是否正确


101
请求超时，请重试
检查您的网络情况或者联系客服解决


200
系统错误，请联系客服
检查您的接口地址填写是否正确，其他问题联系客服解决


203
暂不支持当前接口



204
请求异常：无该接口权限，请前往官网开通正式接口。
前往官网开通该接口权限，http://www.vv-tool.com


204
请求异常：API调用次数不足,请前往网站兑换次数。
前往官网充值该接口调用次数


401
身份凭证无效
在请求头中添加access_token



接口名称： 获取查询次数

接口地址：http://api.vv-tool.com/tool/accounts/api-detail

请求方式： GET

返回格式： JSON

接口备注： 查询次数

频率限制： 10次/秒


请求参数





参数
类型
必选
描述




name
string
是
searchpush



返回数据说明





参数
类型
描述




surplus_num
int
剩余次数



返回数据


{
    "code":0,
    "msg":"成功",
    "msec":8,
    "time":1571407122,
    "data":{
        "surplus_num":4995
    }
}

拉模式和推模式区别
拉模式（定时轮询访问接口获取数据）
数据更新频率低，则大多数的数据请求时无效的
在线用户数量多，则服务端的查询负载很高
定时轮询拉取，无法满足时效性要求
推模式（向客户端进行数据的推送）
仅在数据更新时，才有推送
需要维护大量的在线长连接
数据更新后，可以立即推送
基于WebSocket协议做推送
浏览器支持的socket编程，轻松维持服务端的长连接
基于TCP协议之上的高层协议，无需开发者关心通讯细节
提供了高度抽象的编程接口，业务开发成本较低
WebSocket协议的交互流程
客户端首先发起一个Http请求到服务端，请求的特殊之处，在于在请求里面带了一个upgrade的字段，告诉服务端，我想生成一个websocket的协议，服务端收到请求后，会给客户端一个握手的确认，返回一个switching， 意思允许客户端向websocket协议转换，完成这个协商之后，客户端与服务端之间的底层TCP协议是没有中断的，接下来，客户端可以向服务端发起一个基于websocket协议的消息，服务端也可以主动向客户端发起websocket协议的消息，websocket协议里面通讯的单位就叫message。
————————————————
服务端技术选型与考虑
NodeJs
单线程模型（尽管可以多进程），推送性能有限
C/C++
TCP通讯、WebSocket协议实现成本高
Go
多线程，基于协程模型并发
Go语言属于编译型语言，运行速度并不慢
成熟的WebSocket标准库，无需造轮子
基于Go实现WebSocket服务端
用Go语言对WebSocket做一个简单的服务端实现，以及HTML页面进行调试，并对WebSocket封装，这里就直接给出代码了。
WebSocket服务端
package main
 
import (
		"net/http"
		"github.com/gorilla/websocket"
		"github.com/myproject/gowebsocket/impl"
		"time"
		)
var(
	upgrader = websocket.Upgrader{
		// 允许跨域
		CheckOrigin:func(r *http.Request) bool{
			return true
		},
	}
)
 
func wsHandler(w http.ResponseWriter , r *http.Request){
	//	w.Write([]byte("hello"))
	var(
		wsConn *websocket.Conn
		err error
		conn *impl.Connection
		data []byte
	)
	// 完成ws协议的握手操作
	// Upgrade:websocket
	if wsConn , err = upgrader.Upgrade(w,r,nil); err != nil{
		return 
	}
 
	if conn , err = impl.InitConnection(wsConn); err != nil{
		goto ERR
	}
 
	// 启动线程，不断发消息
	go func(){
		var (err error)
		for{
			if err = conn.WriteMessage([]byte("heartbeat"));err != nil{
				return 
			}
			time.Sleep(1*time.Second)
		}
	}()
 
	for {
		if data , err = conn.ReadMessage();err != nil{
			goto ERR
		}
		if err = conn.WriteMessage(data);err !=nil{
			goto ERR
		}
	}
 
	ERR:
		conn.Close()
 
}
 
func main(){
 
	http.HandleFunc("/ws",wsHandler)
	http.ListenAndServe("0.0.0.0:7777",nil)
}
前端页面
<!DOCTYPE html>
<html>
<head>
	<title>go websocket</title>
	<meta charset="utf-8" />  
</head>
<body>
	<script type="text/javascript">
		var wsUri ="ws://127.0.0.1:7777/ws"; 
	    var output;  
	    
	    function init() { 
	        output = document.getElementById("output"); 
	        testWebSocket(); 
	    }  
	 
	    function testWebSocket() { 
	        websocket = new WebSocket(wsUri); 
	        websocket.onopen = function(evt) { 
	            onOpen(evt) 
	        }; 
	        websocket.onclose = function(evt) { 
	            onClose(evt) 
	        }; 
	        websocket.onmessage = function(evt) { 
	            onMessage(evt) 
	        }; 
	        websocket.onerror = function(evt) { 
	            onError(evt) 
	        }; 
	    }  
	 
	    function onOpen(evt) { 
	        writeToScreen("CONNECTED"); 
	       // doSend("WebSocket rocks"); 
	    }  
	 
	    function onClose(evt) { 
	        writeToScreen("DISCONNECTED"); 
	    }  
	 
	    function onMessage(evt) { 
	        writeToScreen('<span style="color: blue;">RESPONSE: '+ evt.data+'</span>'); 
	       // websocket.close(); 
	    }  
	 
	    function onError(evt) { 
	        writeToScreen('<span style="color: red;">ERROR:</span> '+ evt.data); 
	    }  
	 
	    function doSend(message) { 
	        writeToScreen("SENT: " + message);  
	        websocket.send(message); 
	    }  
	 
	    function writeToScreen(message) { 
	        var pre = document.createElement("p"); 
	        pre.style.wordWrap = "break-word"; 
	        pre.innerHTML = message; 
	        output.appendChild(pre); 
	    }  
	 
	    window.addEventListener("load", init, false);  
	    function sendBtnClick(){
	    	var msg = document.getElementById("input").value;
	    	doSend(msg);
	    	document.getElementById("input").value = '';
	    }
	    function closeBtnClick(){
	    	websocket.close(); 
	    }
	</script>
	<h2>WebSocket Test</h2>  
	<input type="text" id="input"></input>
	<button onclick="sendBtnClick()" >send</button>
	<button onclick="closeBtnClick()" >close</button>
	<div id="output"></div> 	
	
</body>
</html>
封装WebSocket
package impl
 
import (
		"github.com/gorilla/websocket"
		"sync"
		"errors"
		)
 
type Connection struct{
	wsConnect *websocket.Conn
	inChan chan []byte
	outChan chan []byte
	closeChan chan byte
 
	mutex sync.Mutex  // 对closeChan关闭上锁
	isClosed bool  // 防止closeChan被关闭多次
}
 
func InitConnection(wsConn *websocket.Conn)(conn *Connection ,err error){
	conn = &Connection{
		wsConnect:wsConn,
		inChan: make(chan []byte,1000),
		outChan: make(chan []byte,1000),
		closeChan: make(chan byte,1),
 
	}
	// 启动读协程
	go conn.readLoop();
	// 启动写协程
	go conn.writeLoop();
	return
}
 
func (conn *Connection)ReadMessage()(data []byte , err error){
	
	select{
	case data = <- conn.inChan:
	case <- conn.closeChan:
		err = errors.New("connection is closeed")
	}
	return 
}
 
func (conn *Connection)WriteMessage(data []byte)(err error){
	
	select{
	case conn.outChan <- data:
	case <- conn.closeChan:
		err = errors.New("connection is closeed")
	}
	return 
}
 
func (conn *Connection)Close(){
	// 线程安全，可多次调用
	conn.wsConnect.Close()
	// 利用标记，让closeChan只关闭一次
	conn.mutex.Lock()
	if !conn.isClosed {
		close(conn.closeChan)
		conn.isClosed = true 
	}
	conn.mutex.Unlock()
}
 
// 内部实现
func (conn *Connection)readLoop(){
	var(
		data []byte
		err error
		)
	for{
		if _, data , err = conn.wsConnect.ReadMessage(); err != nil{
			goto ERR
		}
//阻塞在这里，等待inChan有空闲位置
		select{
			case conn.inChan <- data:
			case <- conn.closeChan:		// closeChan 感知 conn断开
				goto ERR
		}
		
	}
 
	ERR:
		conn.Close()
}
 
func (conn *Connection)writeLoop(){
	var(
		data []byte
		err error
		)
 
	for{
		select{
			case data= <- conn.outChan:
			case <- conn.closeChan:
				goto ERR
		}
		if err = conn.wsConnect.WriteMessage(websocket.TextMessage , data); err != nil{
			goto ERR
		}
	}
 
	ERR:
		conn.Close()
 
}
千万级弹幕系统的架构设计
技术难点
内核瓶颈
推送量大：100W在线 * 10条/每秒 = 1000W条/秒
内核瓶颈：linux内核发送TCP的极限包频 ≈ 100W/秒
锁瓶颈
需要维护在线用户集合（100W用户在线），通常是一个字典结构
推送消息即遍历整个集合，顺序发送消息，耗时极长 
推送期间，客户端仍旧正常的上下线，集合面临不停的修改，修改需要遍历，所以集合需要上锁
CPU瓶颈
浏览器与服务端之间一般采用的是JSon格式去通讯
Json编码非常耗费CPU资源
向100W在线推送一次，则需100W次Json Encode
优化方案
内核瓶颈
减少网络小包的发送，我们将网络上几百字节定义成网络的小包了，小包的问题是对内核和网络的中间设备造成处理的压力。方案是将一秒内N条消息合并成1条消息，合并后，每秒推送数等于在线连接数。
锁瓶颈
大锁拆小锁，将长连接打散到多个集合中去，每个集合都有自己的锁，多线程并发推送集合，线程之间推送的集合不同，所以没有锁的竞争关系，避免锁竞争。
读写锁取代互斥锁，多个推送任务可以并发遍历相同集合
CPU瓶颈
减少重复计算，Json编码前置，1次消息编码+100W次推送，消息合并前置，N条消息合并后，只需要编码一次。
集群
部署多个节点，通过负载均衡，把连接打散到多个 服务器上，但推送消息的时候，不知道哪个直播间在哪个节点上，最常用的方式是将消息广播给所有的网关节点，此时就需要做一个逻辑集群。
逻辑集群
基于Http2协议向gateway集群分发消息（Http2支持连接复用，用作RPC性能更佳，即在单个连接上可以做高吞吐的请求应答处理）
基于Http1协议对外提供推送API（Http1更加普及，对业务方更加友好）
整体分布式架构图如下：
任何业务方通过Http接口调用到逻辑集群，逻辑集群把消息广播给所有网关，各个网关各自将消息推送给在线的连接即可。
本文讲解了开发消息推送服务的难点与解决方案的大体思路，按照整个理论流程下来，基本能实现一套弹幕消息推送的服务。

本人使用的是3.0.1版本 ，现在已经是4.0版本以后了，版本区别很大方法就不同，4.0是使用的接口。。
import org.apache.commons.httpclient.HttpClient
import org.apache.commons.httpclient.methods.PostMethod
import org.apache.commons.httpclient.*;
import org.apache.commons.httpclient.methods.GetMethod;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.params.HttpClientParams;
import org.apache.commons.httpclient.params.HttpMethodParams;


public static String setUevent(url){
HttpClient       httpClient   = new HttpClient();
PostMethod method  = new PostMethod(url)
method.getParams().setParameter(HttpMethodParams.HTTP_CONTENT_CHARSET,"utf-8");//防止中文乱码

这里有两种传参方法
1、
method.addParameter("user", 'user');//使用string类型不是的记得.toString()
method.addParameter("pass",  'pass');
httpClient.executeMethod(method);
method.releaseConnection();//记得关闭
2、
NameValuePair[] data =[
new NameValuePair("CLIENT_ID", uevent.clientcmy.id.toString()),
new NameValuePair("CONTENT",  uevent.content.toString())
]
method.setRequestBody(data);
try {
			httpClient.getParams().setContentCharset("GBK");
			// 执行postMethod
			int statusCode = httpClient.executeMethod(method);
			// HttpClient对于要求接受后继服务的请求，象POST和PUT等不能自动处理转发
			// 301或者302
			//    if (statusCode == HttpStatus.SC_MOVED_PERMANENTLY
			//      || statusCode == HttpStatus.SC_MOVED_TEMPORARILY) {
			//     String responseBody = postMethod.getResponseBodyAsString();
			//     log.info(responseBody);
			//     if (responseBody != null) {
			//      if ("succ".equals(responseBody)) { // 发送成功
			//
			//       isSended = true;
			//      }
			//     }
			//    }

			if (statusCode == 200) {
				strResponse = method.getResponseBodyAsString();
			}
		} catch (IllegalArgumentException e) {
			e.printStackTrace();
		} catch (HttpException e) {
			// 发生致命的异常，可能是协议不对或者返回的内容有问题

			e.printStackTrace();
		} catch (IOException e) {
			// 发生网络异常
			e.printStackTrace();
		} finally {
			// 释放连接
			method.releaseConnection();
		}


}

参考：https://www.imooc.com/learn/1025

拉模式和推模式区别

拉模式（定时轮询访问接口获取数据）

数据更新频率低，则大多数的数据请求时无效的
	
在线用户数量多，则服务端的查询负载很高
	
定时轮询拉取，无法满足时效性要求
推模式（向客户端进行数据的推送）

仅在数据更新时，才有推送
	
需要维护大量的在线长连接
	
数据更新后，可以立即推送
基于WebSocket协议做推送

浏览器支持的socket编程，轻松维持服务端的长连接
	
基于TCP协议之上的高层协议，无需开发者关心通讯细节
	
提供了高度抽象的编程接口，业务开发成本较低
WebSocket协议的交互流程



 

客户端首先发起一个Http请求到服务端，请求的特殊之处，在于在请求里面带了一个upgrade的字段，告诉服务端，我想生成一个websocket的协议，服务端收到请求后，会给客户端一个握手的确认，返回一个switching， 意思允许客户端向websocket协议转换，完成这个协商之后，客户端与服务端之间的底层TCP协议是没有中断的，接下来，客户端可以向服务端发起一个基于websocket协议的消息，服务端也可以主动向客户端发起websocket协议的消息，websocket协议里面通讯的单位就叫message。

传输协议原理

协议升级后，继续复用Http协议的底层socket完成后续通讯
	
message底层会被切分成多个frame帧进行传输，从协议层面不能传输一个大包，只能切成一个个小包传输
	
编程时，只需操作message，无需关心frame（属于协议和类库自身去操作的）
	
框架底层完成TCP网络I/O，WebSocket协议的解析，开发者无需关心
服务端技术选型与考虑

NodeJs

单线程模型（尽管可以多进程），推送性能有限
C/C++

TCP通讯、WebSocket协议实现成本高
Go

多线程，基于协程模型并发
	
Go语言属于编译型语言，运行速度并不慢
	
成熟的WebSocket标准库，无需造轮子
基于Go实现WebSocket服务端

用Go语言对WebSocket做一个简单的服务端实现，以及HTML页面进行调试，并对WebSocket封装，这里就直接给出代码了。

WebSocket服务端

package main
 
import (
		"net/http"
		"github.com/gorilla/websocket"
		"github.com/myproject/gowebsocket/impl"
		"time"
		)
var(
	upgrader = websocket.Upgrader{
		// 允许跨域
		CheckOrigin:func(r *http.Request) bool{
			return true
		},
	}
)
 
func wsHandler(w http.ResponseWriter , r *http.Request){
	//	w.Write([]byte("hello"))
	var(
		wsConn *websocket.Conn
		err error
		conn *impl.Connection
		data []byte
	)
	// 完成ws协议的握手操作
	// Upgrade:websocket
	if wsConn , err = upgrader.Upgrade(w,r,nil); err != nil{
		return 
	}
 
	if conn , err = impl.InitConnection(wsConn); err != nil{
		goto ERR
	}
 
	// 启动线程，不断发消息
	go func(){
		var (err error)
		for{
			if err = conn.WriteMessage([]byte("heartbeat"));err != nil{
				return 
			}
			time.Sleep(1*time.Second)
		}
	}()
 
	for {
		if data , err = conn.ReadMessage();err != nil{
			goto ERR
		}
		if err = conn.WriteMessage(data);err !=nil{
			goto ERR
		}
	}
 
	ERR:
		conn.Close()
 
}
 
func main(){
 
	http.HandleFunc("/ws",wsHandler)
	http.ListenAndServe("0.0.0.0:7777",nil)
}

前端页面

<!DOCTYPE html>
<html>
<head>
	<title>go websocket</title>
	<meta charset="utf-8" />  
</head>
<body>
	<script type="text/javascript">
		var wsUri ="ws://127.0.0.1:7777/ws"; 
	    var output;  
	    
	    function init() { 
	        output = document.getElementById("output"); 
	        testWebSocket(); 
	    }  
	 
	    function testWebSocket() { 
	        websocket = new WebSocket(wsUri); 
	        websocket.onopen = function(evt) { 
	            onOpen(evt) 
	        }; 
	        websocket.onclose = function(evt) { 
	            onClose(evt) 
	        }; 
	        websocket.onmessage = function(evt) { 
	            onMessage(evt) 
	        }; 
	        websocket.onerror = function(evt) { 
	            onError(evt) 
	        }; 
	    }  
	 
	    function onOpen(evt) { 
	        writeToScreen("CONNECTED"); 
	       // doSend("WebSocket rocks"); 
	    }  
	 
	    function onClose(evt) { 
	        writeToScreen("DISCONNECTED"); 
	    }  
	 
	    function onMessage(evt) { 
	        writeToScreen('<span style="color: blue;">RESPONSE: '+ evt.data+'</span>'); 
	       // websocket.close(); 
	    }  
	 
	    function onError(evt) { 
	        writeToScreen('<span style="color: red;">ERROR:</span> '+ evt.data); 
	    }  
	 
	    function doSend(message) { 
	        writeToScreen("SENT: " + message);  
	        websocket.send(message); 
	    }  
	 
	    function writeToScreen(message) { 
	        var pre = document.createElement("p"); 
	        pre.style.wordWrap = "break-word"; 
	        pre.innerHTML = message; 
	        output.appendChild(pre); 
	    }  
	 
	    window.addEventListener("load", init, false);  
	    function sendBtnClick(){
	    	var msg = document.getElementById("input").value;
	    	doSend(msg);
	    	document.getElementById("input").value = '';
	    }
	    function closeBtnClick(){
	    	websocket.close(); 
	    }
	</script>
	<h2>WebSocket Test</h2>  
	<input type="text" id="input"></input>
	<button onclick="sendBtnClick()" >send</button>
	<button onclick="closeBtnClick()" >close</button>
	<div id="output"></div> 	
	
</body>
</html>

封装WebSocket

package impl
 
import (
		"github.com/gorilla/websocket"
		"sync"
		"errors"
		)
 
type Connection struct{
	wsConnect *websocket.Conn
	inChan chan []byte
	outChan chan []byte
	closeChan chan byte
 
	mutex sync.Mutex  // 对closeChan关闭上锁
	isClosed bool  // 防止closeChan被关闭多次
}
 
func InitConnection(wsConn *websocket.Conn)(conn *Connection ,err error){
	conn = &Connection{
		wsConnect:wsConn,
		inChan: make(chan []byte,1000),
		outChan: make(chan []byte,1000),
		closeChan: make(chan byte,1),
 
	}
	// 启动读协程
	go conn.readLoop();
	// 启动写协程
	go conn.writeLoop();
	return
}
 
func (conn *Connection)ReadMessage()(data []byte , err error){
	
	select{
	case data = <- conn.inChan:
	case <- conn.closeChan:
		err = errors.New("connection is closeed")
	}
	return 
}
 
func (conn *Connection)WriteMessage(data []byte)(err error){
	
	select{
	case conn.outChan <- data:
	case <- conn.closeChan:
		err = errors.New("connection is closeed")
	}
	return 
}
 
func (conn *Connection)Close(){
	// 线程安全，可多次调用
	conn.wsConnect.Close()
	// 利用标记，让closeChan只关闭一次
	conn.mutex.Lock()
	if !conn.isClosed {
		close(conn.closeChan)
		conn.isClosed = true 
	}
	conn.mutex.Unlock()
}
 
// 内部实现
func (conn *Connection)readLoop(){
	var(
		data []byte
		err error
		)
	for{
		if _, data , err = conn.wsConnect.ReadMessage(); err != nil{
			goto ERR
		}
//阻塞在这里，等待inChan有空闲位置
		select{
			case conn.inChan <- data:
			case <- conn.closeChan:		// closeChan 感知 conn断开
				goto ERR
		}
		
	}
 
	ERR:
		conn.Close()
}
 
func (conn *Connection)writeLoop(){
	var(
		data []byte
		err error
		)
 
	for{
		select{
			case data= <- conn.outChan:
			case <- conn.closeChan:
				goto ERR
		}
		if err = conn.wsConnect.WriteMessage(websocket.TextMessage , data); err != nil{
			goto ERR
		}
	}
 
	ERR:
		conn.Close()
 
}

千万级弹幕系统的架构设计

技术难点

内核瓶颈

推送量大：100W在线 * 10条/每秒 = 1000W条/秒
	
内核瓶颈：linux内核发送TCP的极限包频 ≈ 100W/秒
锁瓶颈

需要维护在线用户集合（100W用户在线），通常是一个字典结构
	
推送消息即遍历整个集合，顺序发送消息，耗时极长 
	
推送期间，客户端仍旧正常的上下线，集合面临不停的修改，修改需要遍历，所以集合需要上锁
CPU瓶颈

浏览器与服务端之间一般采用的是JSon格式去通讯
	
Json编码非常耗费CPU资源
	
向100W在线推送一次，则需100W次Json Encode
优化方案

内核瓶颈

减少网络小包的发送，我们将网络上几百字节定义成网络的小包了，小包的问题是对内核和网络的中间设备造成处理的压力。方案是将一秒内N条消息合并成1条消息，合并后，每秒推送数等于在线连接数。
锁瓶颈

大锁拆小锁，将长连接打散到多个集合中去，每个集合都有自己的锁，多线程并发推送集合，线程之间推送的集合不同，所以没有锁的竞争关系，避免锁竞争。
	
读写锁取代互斥锁，多个推送任务可以并发遍历相同集合
CPU瓶颈

减少重复计算，Json编码前置，1次消息编码+100W次推送，消息合并前置，N条消息合并后，只需要编码一次。
单机架构



 

最外层是在线的长连接，连接到服务端后，打散到多个集合里面存储，我们要发送的消息呢，通过打包后，经过json编码，被多个线程或协程分发到多个集合中去，最终推给了所有的在线连接。

单机瓶颈

维护海量长连接，会花费不少内存
	
消息推送的瞬时，消耗大量的CPU
	
消息推送的瞬时带宽高达400-600Mb（4-6Gbits），需要用到万兆网卡，是主要瓶颈
集群

部署多个节点，通过负载均衡，把连接打散到多个 服务器上，但推送消息的时候，不知道哪个直播间在哪个节点上，最常用的方式是将消息广播给所有的网关节点，此时就需要做一个逻辑集群。

逻辑集群

基于Http2协议向gateway集群分发消息（Http2支持连接复用，用作RPC性能更佳，即在单个连接上可以做高吞吐的请求应答处理）
	
基于Http1协议对外提供推送API（Http1更加普及，对业务方更加友好）
整体分布式架构图如下：



 

任何业务方通过Http接口调用到逻辑集群，逻辑集群把消息广播给所有网关，各个网关各自将消息推送给在线的连接即可。

本文讲解了开发消息推送服务的难点与解决方案的大体思路，按照整个理论流程下来，基本能实现一套弹幕消息推送的服务。