每秒100w秒杀架构day03-526互联

第3章 100W请求秒杀架构体系-程序隔离和并

发限流

目标1：抢单程序隔离实现

目标2：WebSocket

知识学习

目标3：Sentinel

限流讲解

目标4：LVS+Nginx集群讲解

1 用户下单

商品分为热点商品抢单和非热点商品抢单，因此此系统中抢单模式并非一种。

1.1 抢单分析

如上图，用户登录后经过nginx，进行抢单，此时会先判断商品是否是热点商品，如果是非热点商品，

则直接调用订单系统进行下单操作，如果是热点商品，则向Kafka生产消息进行排队下单，订单系统会

订阅排队下单信息，这样可以降低服务器所直接承受的抢单压力，这种操作也叫队列削峰。

1.2 非热点商品抢单

我们在订单系统中实现非热点商品抢单操作，非热点商品只用在订单系统中实现抢单即可，但抢单的时

候要注意这么几个问题：

1.先递减库存

2.库存递减成功后，执行下单

3.下单失败，需要实现分布式事务

4.下单成功后，要记录用户抢单信息，在24小时内不允许再抢该商品

5.抢单中，有可能存在抢购的商品正好变成了热点商品，此时应该走排队的方式抢单，否则商品数量会发生

不精准问题

1.2.1 库存递减

库存递减我们需要

这里我们需要控制数据的原子性，因此不能在内存中进行操作，需要用SQL语句在数据库中执行。

1)库存递减

修改 seckill-goods 的 com.seckill.goods.service.SkuService 添加库存递减方法，代码如下：

修改 seckill-goods 的 com.seckill.goods.service.impl.SkuServiceImpl 添加库存递减实现方

法，代码如下：

/***

* 递减库存

* @param id

* @param coun

* @return

@Update("U

PDATE tb_sku SET seckill_num=seckill_num-#{count} WHERE id=#{id} AND

islock=1

int dcou

(

Param("count") Integer count);

/***

* 库存递减

* @param id

* @param count

* @return

int dcount(String id, Integer count);

/***

* 库存递减

* @param id

* @param count

* @return

@Override

public int dcount(String id, Integer count) {

int dcount = skuMapper.dcount(id,count);

if(dcount<=0){

//递减失败,查询状态

Sku sku = skuMapper.selectByPrimaryKey(id);

if(sku.getIslock()==2){

return 205;

}

if(sku.getSeckillNum()<count){

return 405;

}

return dcount;

}

修改 com.seckill.goods.controller.SkuController 添加库存递减调用方法，代码如下：

2)SkuFeign配置

修改 com.seckill.goods.feign.SkuFeign ,添加库存递减方法，代码如下：

1.2.2 抢单实现

当库存递减成功后，需要给用户直接下单，如果递减不成功，会出现商品变成热卖商品的现象，我们需

要向Kafka发送队列数据，所以需要引入Kafka配置。

bootstrap.yml配置kafka：

/***

* Sku数量递减

* @return

@PutMapping(v

{

}

)

public Result

(

"id")String

id,@PathVaria

(

)

{

int co

e.dcount(id,count);

String

;

Sku

;

swit

(

)

{

case 1:

message="库存削减成功";

sku = skuService.findById(id);

break;

case 405:

message="库存不足";

break;

case 205:

message="该商品为热点商品";

break;

default:

}

return new Result<Sku>(true,code,message,sku);

}

/***

* Sku数量递减

* @return

@PutMapping(value = "/sku/dcount/{id}/{count}" )

Result<Sku> dcount(@PathVariable(value = "id")String id, @PathVariable(value =

"count")Integer count);

kafka:

producer:

修改 seckill-order 的 com.seckill.order.service.impl.OrderServiceImpl 的add方法，代码如

下：

acks: all #acks:消息的确认机制，默认值是0， acks=0：如果设置为0，生产者不会等待

kafka的响应。 acks=1：这个配置意味着kafka会把这条消息写到本地日志文件中，但是不会等待集群中

其他机器的成功响应。 acks=all：这个配置意味着leader会等待所有的follower同步完成。这个确保

消息不会丢失，除非kafka集群中所有机器挂掉。这是最强的可用性保证。

retries: 0 #发送失败重试次数，配置为大于0的值的话，客户端会在消息发送失败时重新发送。

batch-size: 16384 #当多条消息需要发送到同一个分区时，生产者会尝试合并网络请求。这会

提高client和生产者的效率。

buffer-memory: 33554432 #即32MB的批处理缓冲区

key-serializer: org.apache.kafka.common.serialization.StringSerializer

value-s

bootstr

如

果

启

动

错

误

，

打

开

级

别

日

志

，

出

现Can't re

的

错

误

，

需

要

在

下

修

改

IP映射即可，

C:\Window

k。

cons

d: test

auto-offset-reset: latest #

（1）earliest:当各分区下有已提交的offset时，从提交的

offset开始消费；无提交的offset时，从头开始消费；（2）latest:当各分区下有已提交的offset

时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据；（3）none：

topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的

offset，则抛出异常

enable-auto-commit: true #如果为true，消费者的偏移量将在后台定期提交。

auto-commit-interval: 1000 #消费者偏移自动提交给Kafka的频率（以毫秒为单位），默

认值为5000

max-poll-records: 5 #一次拉起的条数

key-deserializer: org.apache.kafka.common.serialization.StringDeserializer

value-deserializer:

org.apache.kafka.common.serialization.StringDeserializer

bootstrap-servers: kafka-server:9092

/**

* 增加Order

* @param order

@Override

public int add(Order order){

//1.库存递减

Result<Sku> result = skuFeign.dcount(order.getId(), 1);

//2.递减成功，下单

if(result.getCode()==1){

//商品秒杀价格

Sku sku = result.getData();

order.setName(sku.getName());

order.setPrice(sku.getSeckillPrice());

order.setId("No"+idWorker.nextId());

order.setOrderStatus("0");

order.setPayStatus("0");

order.setConsignStatus("0");

orderMapper.insertSelective(order);

}else if(result.getCode()==205){//3.商品转入了热点商品,排队

//检查用户是否在排队

String key = "SKU_"+order.getSkuId();

修改 com.seckill.order.controller.OrderController 的add方法，代码如下：

String userKey = "USER"+order.getUsername()+"ID"+order.getSkuId();

//如果为true，则表示用户正在排队

Boolean bo = redisTemplate.boundHashOps(key).hasKey(userKey);

if(!bo){

Map<String,String> orderMap = new HashMap<String,String>();

orderMap.put("username",order.getUsername());

orderMap.put("id",order.getSkuId());

//抢单排队

afkaTemplate.send("neworder", JSON.toJSONString(orderMap));

}

turn 202;

}

retu

rn result.getCode();

}

/***

* 新增Order数据

* @return

@GetMapping(value = "/{id}")

public Result add(@RequestHeader(value = "Authorization")String authorization,//

令牌

@PathVariable(value = "id")String id) //商品ID

{

//解析令牌

Map<String,Object> tokenMap = JwtTokenUtil.parseToken(authorization);

Order order = new Order();

order.setCreateTime(new Date());

order.setUpdateTime(order.getCreateTime());

order.setSkuId(id);

order.setUsername(tokenMap.get("username").toString());

//调用OrderService实现添加Order

int code = orderService.add(order);

String message = "";

switch (code){

case 405:

message="库存不足";

break;

case 200:

message="抢购成功";

break;

case 202:

message="正在排队";

break;

case 1:

message="抢单成功";

break;

default:

}

return new Result(true, StatusCode.OK,message);

}

1.2.3 抢单测试

编写lua脚本控制抢单，当用户处于已登录状态，则执行下单,创建脚本 seckill-order-add.lua ，脚本

如下：

在nginx.conf中添加api以及抢单请求路径的路由，配置如下：

注意如果是POST提交，Lua脚本中需要读取请求体数据：

ngx.header.content_type="application/json;charset=utf8"

--引入json库

local cjson =

require "cjson"

--引入jwt模块

local jwtt

quire "token"

--获取请求头

中

的

令

牌

数

据

local au

header = ngx.var.http_Authorization

--调用令牌

校

验

local result = jwttoken.check(auth_header)

--如果code==200表示令牌校验通过

if result.code==200 then

--获取id

local uri_args = ngx.req.get_uri_args()

local id = uri_args["id"]

--拼接url

local url = "/api/order/"..id

--执行请求

ngx.exec(url)

else

-- 输出结果

ngx.say(cjson.encode(result))

ngx.exit(result.code)

end

#抢单

location /lua/order/add {

content_by_lua_file /usr/local/openresty/nginx/lua/seckill-order-add.lua;

}

#微服务网关

location /api/ {

proxy_pass http://192.168.1.5:8001;

}

ngx.req.read_body()

ngx.req.get_post_args()

1.3 热点商品抢单

上面我们完成了非热点商品抢单，接着我们实现以下热点商品抢单。热点商品和非热点商品不一样，热

点商品已经隔离出来，在Redis缓存中，并且热点商品抢单要实现高效率操作而且还能抗压，Nginx的并

发能力远远超越tomcat，因此热点商品抢单我们可以使用Lua+Nginx。

1.3.1 抢单流程分析

用户进入抢单流程，通过Lua脚本判断令牌是否有效，如果有效，则进入抢单环节，抢单环节执行过程

我们做一个分析：

1.3.2 Lua实现Redis操作

判断用户是否在24小时内抢购过该商品，我们可以将用户抢单信息存入到Redis缓存中，定时24小时过

期即可，此时需要在Lua里面实现Redis集群操作，需要第三方库的支持 lua-resty-redis-cluster 。

我们需要安装 lua-resty-redis-cluster ，下载地址： <https://github.com/cuiweixie/lua

resty-redis-cluster> ，下载该文件配置后即可实现Redis集群操作。

5.3.2.1 配置lua-resty-redis-cluster

1)lua-resty-redis-cluster配置

将下载好的文件上传到服务器的 `/usr/local/openresty 目录下，并解压，我们只需要用到包中2个

文件rediscluster.lua和 redis_slot.c 。

1.判断该商品用户是否在24小时内购买过

2.如果购买了，直接提示用户24小时内无法购买

3.如果用户没有购买过该商品，则判断该商品是否属于热点商品

4.如果是非热点商品，则走非热点商品抢单流程

5.如果是热点商品，则走热点商品抢单流程

6.判断该商品用户是否已经排队，如果没有排队，则进入排队，如果已经排队，则提示用户正在排队

7.下单过程交给订单系统，订单系统通过队列订阅读取用户下单信息，并进行下单

将 lua-resty-redis-cluster/lib/redis_slot.c 拷贝到 openresty/lualib 目录下，将 lua

resty-redis-cluster/lib/resty/rediscluster.lua 拷贝到 openresty/lualib/resty 目录下。

拷贝redis_slot.c：

拷贝rediscluster.lua

：

编译：

编译的时候有可能会发生如下错误

错误解决:

这里我们可以选择第二种方式解决,再次进行编译就没问题了。

2)指令配置

lua-resty-redis-cluster 中有部分redis指令并未开放，我们可以手动修改，开放相关指令,我们这里

开放过期指令，因为后面会用到该指令。

修改 /usr/local/openresty/lualib/resty/rediscluster.lua 文件，添加相关指令，如下图：

cd /usr/local/openresty/lua-resty-redis-cluster-master/lib/

cp redis_slot.c /usr/local/openresty/lualib/

cd /usr/

local/openresty/lua-resty-redis-cluster-master/lib/resty

cp rediscluster.lua /usr/local/openresty/lualib/resty/

cd /usr/local/openresty/lualib

gcc redis_slot.c -fPIC -shared -o libredis_slot.so

解决：应该是lua版本不对，自带的lua应该不好使

方式一：删除自带的lua，一般是/usr/lua和/usr/luac ,删除这两个文件

方式二：yum install lua-devel 下载一个依赖

方式三:自己重新再lua官网下载一个lua，重新安装一个lua（这个很好使）

1.3.2.2 操作Redis集群实现

以后别的地方也有可能会用到redis，我们可以写个工具类 redis-cluster.lua ，实现redis的操作，这

里主要实现了根据key获取缓存数据、根据key设置缓存过期时间、根据key从hash类型中获取数据、往

hash类型中添加数据，代码如下：

--redis连接配置

local config = {

name = "test",

serv_list = {

{ip="192.168.211.137", port = 7001},

{ip="192.168.211.137", port = 7002},

{ip="192.168.211.137", port = 7003},

{ip="192.168.211.137", port = 7004},

{ip="192.168.211.137", port = 7005},

{ip="192.168.211.137", port = 7006},

idle_timeout = 1000,

pool_size = 10000,

}

--引入redis集群配置

local redis_cluster = require "resty.rediscluster"

--定义一个对象

local lredis = {}

--根据key查询

function lredis.get(key)

--创建链接

local red = redis_cluster:new(config)

red:init_pipeline()

--根据key获取数据

red:get(key)

local rresult = red:commit_pipeline()

--关闭链接

red:close()

return rresult

end

--添加带过期的数据

function lredis.setexp(key,value,time)

--创建链接

local red = redis_cluster:new(config)

red:i

nit_pipeline()

-添

加

，

同

时

设

置

过

期时间

(

)

red:expire(key,time)

local rresult = red:commit_pipeline()

end

--根据key查询hash

function lredis.hget(key1,key2)

--创建链接

local red = redis_cluster:new(config)

red:init_pipeline()

--根据key获取数据

red:hmget(key1,key2)

local rresult = red:commit_pipeline()

--关闭链接

red:close()

return rresult[1]

end

--hash数据添加

function lredis.hset(key1,key2,value)

--创建链接

local red = redis_cluster:new(config)

red:init_pipeline()

--添加hash数据

red:hmset(key1,key2,value)

local rresult = red:commit_pipeline()

--关闭链接

red:close()

return rresult

end

--hash中指定的key自增

function lredis.hincrby(key1,key2,value)

--创建链接

local red = redis_cluster:new(config)

red:init_pipeline()

--添加hash数据

red:hincrby(key1,key2,value)

local rresult = red:commit_pipeline()

我们接着来测试一次集群操作，修改

nginx.conf ，配置一个 location 节点，如下：

测试效果如下：

1.3.3 Lua实现Kafka操作

用户抢单的时候，如果是热点商品，这时候需要实现用户排队，用户排队我们需要向kafka发送抢单信

息，因此需要使用Lua脚本操作kafka，我们需要依赖 lua-restry-kafka 库，该库我们也已经配置使用

过了，所以这里无需再配置了。

库文件使用：https://github.com/doujiang24/lua-resty-kafka

以后使用kafka的地方也有可能会有很多，所以针对kafka我们也可以单独抽取出一个配置脚本，创建一

个脚本名字叫'kafka.lua'，用于配置kafka的操作，代码如下：

--关闭链接

red:close()

return rresult

end

return lredis

#redis

location

/test/redis {

content_by_lua '

ngx.header.content_type="application/json;charset=utf8"

--引入redis

local rredis = require "redis-cluster"

--从redis中查询hash类型数据

local sku = rredis.hget("SKU_S1235433012716498944","num")[1]

ngx.say(sku)

}

--创建对象

local kafka={}

--kafka依赖库

local client = require "resty.kafka.client"

local producer = require "resty.kafka.producer"

--配置kafka的链接地址

local broker_list = {

{ host = "192.168.211.137", port = 9092 }

}

--发送消息

--queuename:队列名字

--content:发送的内容

编写一段代码向

kafka发送信息，修改 nginx.conf ,添加如下代码：

测试效果如下：

1.3.4 抢单实现

function kafka.send(queuename,content)

--创建生产者

local pro = producer:new(broker_list,{ producer_type="async"})

--发送消息

local offset, err = pro:send(queuename, nil, content)

--返回结果

return offset

end

return kafka

#kafka

location /test/kafka {

content_by_lua '

ngx.header.content_type="application/json;charset=utf8"

--引入kafka

local kafka = require "kafka"

--发送消息

local offset = kafka.send("demo","hello")

ngx.say(offset)

}

抢单这里分为2部分，首先需要向Kafka发送抢单信息实现排队，排队后，订单系统订阅抢单信息实现下

单操作，所有的数据操作一律在Redis中完成，降低程序对服务器的压力。

1.3.4.1 排队

排队抢单需要引入redis和kafka，我们的实现思路如下：

创建 seckill-order-add.lua ,实现代码如下：

1.校验用户令牌，如果不通过直接结束程序提示用户

2.令牌校验通过，从Redis中获取用户在24小时内是抢购过该商品，如果抢购过直接结束程序并提示用户

3.如果符合购买该商品条件，则校验该商品是否是热点商品，如果不是，直接请求后台下单

4.如果是热点商品，并且库存>0，校验用户是否已经在排队，使用redis的incr自增判断排队次数可以去

除重复排队

5.如果没有排队，则向Kafka发送消息实现排队

ngx.header.content_type="application/json;charset=utf8"

--引入json库

local cjson = require "cjson"

--引入jwt模块

local jwttoken = require "token"

--引入redis

local redis = require "redis-cluster"

--引入kafka

local kafka = require "kafka"

--获取请求头中的令牌数据

local auth_header = ngx.var.http_Authorization

--调用令牌校验

local result = jwttoken.check(auth_header)

--如果code==200表示令牌校验通过

if result.code==200 then

--响应结果

local response = {}

--获取id

local uri_args = ngx.req.get_uri_args()

local id = uri_args["id"]

--判断该商品用户是否已经在指定时间内购买过

local username = result["body"]["payload"]["username"]

local userKey= "USER"..username.."ID"..id

local hasbuy = redis.get(userKey)

--如果没有购买，则判断该商品是否是热点商品

if hasbuy==nil or hasbuy[1]==nil or hasbuy[1]==ngx.null then

--从redis中获取该商品信息

local num = redis.hget("SKU_"..id,"num")[1]

--如果

不是热点商品，则走普通抢单流程

or num==ngx.null then

-拼

接

rl = "/api/order/"..id

-执

行

请

求

ngx.exec(url)

return

else

--热点商品

num = tonumber(num)

--如果有库存，才允许抢单

if num<=0 then

--库存不足，无法排队

response["code"]=405

response["message"]="当前商品库存不足，无法抢购"

ngx.say(cjson.encode(response))

return

else

--递增排队

local incrresult = redis.hincrby("SKU_"..id,userKey,1)

incrresult=tonumber(incrresult)

if incrresult==1 then

--热点数据，发送MQ排队

local userorder = {}

userorder["username"]=username

userorder["id"]=id

--排队抢单

kafka.send("neworder",cjson.encode(userorder))

response["code"]=202

response["message"]="您正在排队抢购该商品"

ngx.say(cjson.encode(response))

return

else

--响应用户正在排队抢购该商品

response["code"]=202

response["message"]="您正在排队抢购该商品"

ngx.say(cjson.encode(response))

return

end

else

--24小时内购买过该商品

response["code"]=415

1.3.4.2 下单实现

创建 com.se

ckill.order.config.KafkaOrderListener ,用于读取排队信息，并调用下单操作，代码

如下：

修改 com.seckill.order.service.OrderService 添加热点数据下单方法，代码如下：

修改 com.seckill.order.service.impl.OrderServiceImpl 添加热点数据下单实现方法，代码如

下：

response["message"]="您24小时内已经抢购了该商品，不能重复抢购"

ngx.say(cjson.encode(response))

return

end

else

-- 输出结果

ngx.say(cjson.encode(result))

ngx.exit(result.code)

end

@Component

public class KafkaOrderListener {

@Autowired

private OrderService orderService;

/***

* 监听消息

* 创建订单

* @param message

@KafkaListener(topics = {"neworder"})

public void receive(String message){

//将消息转成Map

Map<String,String> orderMap = JSON.parseObject(message,Map.class);

//创建订单

orderService.addHotOrder(orderMap);

}

/***

* 热点数据下单

* @param orderMap

void addHotOrder(Map<String, String> orderMap);

/***

* 秒杀下单

* @param orderMap

@Override

public void addHotOrder(Map<String, String> orderMap) {

String id = orderMap.get("id");

String username = orderMap.get("username");

//key

String key = "SKU_" + id;

//用户购买的key

String userKey = "USER" + username + "ID" + id;

if (redisTemplate.hasKey(key)) {

//数量

Integ

Integer.parse

(

sTemplate.boundHashOps(key).get("num").toString());

拥有库存，执行递减操作，需要控制超卖（分布式锁可以解决、或者Redis也可以解决）

(

{

查

询

商

品

Result<Sku> result = skuFeign.findById(id);

Sku sku = result.getData();

Order order = new Order();

order.setCreateTime(new Date());

order.setUpdateTime(order.getCreateTime());

order.setUsername(username);

order.setSkuId(id);

order.setName(sku.getName());

order.setPrice(sku.getSeckillPrice());

order.setId("No" + idWorker.nextId());

order.setOrderStatus("0");

order.setPayStatus("0");

order.setConsignStatus("0");

orderMapper.insertSelective(order);

//库存递减

num--;

if (num == 0) {

//同步数据到数据库，秒杀数量归零

skuFeign.zero(id);

}

//更新数据

Map<String, Object> dataMap = new HashMap<String, Object>();

dataMap.put("num", num);

dataMap.put(userKey, 0);

//存数据

redisTemplate.boundHashOps(key).putAll(dataMap);

}

//记录该商品用户24小时内无法再次购买,测试环境，我们只配置成1分钟

redisTemplate.boundValueOps(userKey).set("");

redisTemplate.expire(userKey, 1, TimeUnit.MINUTES);

}

2 WebSocket（自学）

目前用户抢单操作我们已经完成，无论是非热点商品还是热点商品抢单，抢单完成后，我们应该要通知

用户抢单状态，非热点商品可以直接响应抢单结果，但热点商品目前还没有实现通知响应，通知用户抢

单状态用户可以通过定时向后台发出请求查询实现，但这种短连接方式效率低，会和服务器进行多次通

信，这块我们可以使用长连接websocket实现。

2.1 WebSock

et介绍

WebSocket 是

HTML5 开始提供的一种在单个 TCP 连接上进行全双向通讯的协议。

WebSocket

使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。

在 WebSock

et API 中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连

接，并进行双向数据传输。

现在，很多网站为了实现推送技术，所用的技术都是 Ajax 轮询。轮询是在特定的的时间间隔（如每1

秒），由浏览器对服务器发出HTTP请求，然后由服务器返回最新的数据给客户端的浏览器。这种传统

的模式带来很明显的缺点，即浏览器需要不断的向服务器发出请求，然而HTTP请求可能包含较长的头

部，其中真正有效的数据可能只是很小的一部分，显然这样会浪费很多的带宽等资源。

HTML5 定义的 WebSocket 协议，能更好的节省服务器资源和带宽，并且能够更实时地进行通讯。

浏览器通过 JavaScript 向服务器发出建立 WebSocket 连接的请求，连接建立以后，客户端和服务器端

就可以通过 TCP 连接直接交换数据。

当你获取 Web Socket 连接后，你可以通过 send() 方法来向服务器发送数据，并通过 onmessage 事

件来接收服务器返回的数据。

2.2 Websocket API

创建 WebSocket 对象：

WebSocket属性：

var socket = new WebSocket(url, [protocol] );

属性

描述

socket.readyState

只读属性 readyState 表示连接状态，可以是以下值：0 - 表示连接

尚未建立。1 - 表示连接已建立，可以进行通信。2 - 表示连接正在进

行关闭。3 - 表示连接已经关闭或者连接不能打开。

socket.bufffferedAmount

只读属性 bufffferedAmount 已被 send() 放入正在队列中等待传

输，但是还没有发出的 UTF-8 文本字节数。

事件

事件处理程序

描述

open

socket.onopen

连接建立时触发

message

socket.onmessage

客户端接收服务端数据时触发

error

socket.onerror

通信发生错误时触发

socket.onclose

连接关闭时触发

方法

描述

socket.send()

使用连接发送数据

socket.close()

关闭连接

WebSocket事件：

WebSocket方法：

2.3 WebSocket实例

2.3.1 客户端

WebSocket 协议本质上是一个基于 TCP 的协议。为了建立一个 WebSocket 连接，客户端浏览器首先

要向服务器发起一个 HTTP 请求，这个请求和通常的 HTTP 请求不同，包含了一些附加头信息，其中附

加头信息"Upgrade: WebSocket"表明这是一个申请协议升级的 HTTP 请求，服务器端解析这些附加的

头信息然后产生应答信息返回给客户端，客户端和服务器端的 WebSocket 连接就建立起来了，双方就

可以通过这个连接通道自由的传递信息，并且这个连接会持续存在直到客户端或者服务器端的某一方主

动的关闭连接。

我们按照上面的API实现客户端代码如下：

<!DOCTYPE html>

<html>

<head>

<title>websocket</title>

var socket;

2.3.2 服务端

1)引入依赖包

在 seckill-order 引入如下依赖：

if(typeof(WebSocket) == "undefined") {

console.log("您的浏览器不支持WebSocket");

}else{

console.log("您的浏览器支持WebSocket");

//实现化WebSocket对象，指定要连接的服务器地址与端口建立连接

socket = new WebSocket("ws://localhost:18085/socket/zhangsan");

//打开事件

socket.onopen = function() {

console.log("Socket 已打开");

}

;

获

得

消

息

事

件

ssage = function(msg) {

);

发

现

消

息

进

入

调

后

台

获取

//getCallingList();

$("#msg").append(msg.data+"</br>");

};

//关闭事件

socket.onclose = function() {

console.log("Socket已关闭");

};

//发生了错误事件

socket.onerror = function() {

alert("Socket发生了错误");

}

//关闭连接

function closeWebSocket() {

socket.close();

}

//发送消息

function send() {

var message = document.getElementById('text').value;

socket.send(message);

}

</script>

</head>

<body>

</body>

</html>

2)websocket消息处理

在 seckill-order

服务端，我们可以创建一个类

com.seckill

.order.websocket.WebSocketServer ，并暴露websocket地址出去，代码如下：

<groupId>org.springframework.boot</groupId>

<artifactId>spring-boot-starter-websocket</artifactId>

</dependency>

@Slf4j

@ServerE

ndpoint(value = "/socket/{userid}")

@Component

public class WebSocketServer {

//concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketServer对象。

private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new

CopyOnWriteArraySet<WebSocketServer>();

//与某个客户端的连接会话，需要通过它来给客户端发送数据

private static Map<String,Session> sessions=new HashMap<String,Session>();

//用户唯一标识符

private String userid;

/**

* 连接建立成功调用的方法*/

@OnOpen

public void onOpen(@PathParam("userid") String userid, Session session) {

sessions.put(userid,session);

this.userid=userid;

webSocketSet.add(this); //加入set中

try {

sendMessage("连接成功");

} catch (IOException e) {

log.error("websocket IO异常");

}

/**

* 连接关闭调用的方法

@OnClose

public void onClose() {

webSocketSet.remove(this); //从set中删除

sessions.remove(userid); //移除会话

}

/**

* 收到客户端消息后调用的方法

* @param message 客户端发送过来的消息*/

@OnMessage

public void onMessage(String message) {

log.info("来自客户端的消息:" + message);

//群发消息

for (WebSocketServer item : webSocketSet) {

try {

item.sendMessage(message);

} catch (IOException e) {

e.printStackTrace();

}

/**

* 发

生

异

常

处

理

方

法

* @

* @param error

@OnError

public void onError(Session session, Throwable error) {

error.printStackTrace();

}

/***

* 群发

* @param message

* @throws IOException

public void sendMessage(String message) throws IOException {

for (Map.Entry<String, Session> sessionEntry : sessions.entrySet()) {

sessionEntry.getValue().getBasicRemote().sendText(message);

}

/***

* 给指定用户发送消息

* @param message

* @param userid

* @throws IOException

public void sendMessage(String message,String userid) throws IOException {

sessions.get(userid).getBasicRemote().sendText(message);

}

/**

* 群发自定义消息

* */

public static void sendInfo(String message) throws IOException {

for (WebSocketServer item : webSocketSet) {

try {

item.sendMessage(message);

} catch (IOException e) {

continue;

}

我们编写一个测试类 com.seckill.order.controller.WebSocketController ，用于实现给指定用

户发消息，代码如下：

测试效果如下：

3 Sentinel限流

随着微服务的流行，服务和服务之间的稳定性变得越来越重要。 Sentinel 以流量为切入点，从流量控

制、熔断降级、系统负载保护等多个维度保护服务的稳定性。

3.1 Sentinel介绍

Sentinel 具有以下特征:

@RestController

@RequestMapping(value = "/ws")

public class WebSocketController {

@Autowire

private WebSocketServer webSocketServer;

/***

* 模

拟给指定用户发消息

@GetMapping(value = "/send/{userid}")

public Result sendMessage(@PathVariable(value = "userid")String userid,

String msg) throws IOException {

webSocketServer.sendMessage(msg,userid);

return new Result(true, StatusCode.OK,"发送消息成功@");

}

Sentinel 目前已经针对 Servlet、Dubbo、Spring Boot/Spring Cloud、gRPC 等进行了适配，用户只需

引入相应依赖并进行简单配置即可非常方便地享受

Sentinel 的高可用流量防护能力。Sentinel 还为

Service Mesh 提供了集群流量防护的能力。未来 Sentinel 还会对更多常用框架进行适配。

Sentinel 分为两个部分:

核心库（Java 客户端）不依赖任何框架/库，能够运行于所有 Java 运行时环境，同时对 Dubbo /

Spring Cloud 等框架也有较好的支持。

控制台（Dashboard）基于 Spring Boot 开发，打包后可以直接运行，不需要额外的 Tomcat 等

应用容器。

Sentinel 的关注点在于：

多样化的流量控制

熔断降级

系统负载保护

实时监控和控制台

Sentinel和Hystrix对比：

1.丰富的应用场景： Sentinel 承接了阿里巴巴近 10 年的双十一大促销流量的核心场景，例如秒杀

（即突发流量控制在系统容量可以承受的范围）、消息削峰填谷、实时熔断下游不可用应用等。

2.完备的实时监控： Sentinel 同时提供实时的监控功能。您可以在控制台中看到接入应用的单台机器秒

级数据，甚至 500 台以下规模的集群的汇总运行情况。

3.广泛的开源生态： Sentinel 提供开箱即用的与其它开源框架/库的整合模块，例如与 Spring

Cloud、Dubbo、gRPC 的整合。您只需要引入相应的依赖并进行简单的配置即可快速地接入 Sentinel。

4.完善的 SPI 扩

展点： Sentinel 提供简单易用、完善的 SPI 扩展点。您可以通过实现扩展点，快速

的定制逻辑。例如定制规则管理、适配数据源等。

Sentinel

Hystrix

隔离策略

信号量隔离（并发线程数限流）

线程池隔离/信号量隔离

熔断降级策略

基于响应时间、异常比率、异常数

基于异常比率

常见框架的适

配

Servlet、Spring Cloud、Dubbo、gRPC 等

Servlet、Spring Cloud

Netflflix

规则配置

支持多种数据源

扩展性

多个扩展点

插件的形式

基于注解的支

持

支持

调用链路信息

支持同步调用

不支持

限流

基于 QPS / 并发数，支持基于调用关系的限流

有限支持

控制台

开箱即用，可配置规则、查看秒级监控、机器

发现等

较为简单

系统负载保护

支持

不支持

学习参考地址：<https://github.com/alibaba/spring-cloud-alibaba/wiki/Sentinel

3.2 Sentinel控制台安装

Sentinel 提供一个轻量级的开源控制台，它提供机器发现以及健康情况管理、监控（单机和集群）、规

则管理和推送的功能。

https://github.com/alibaba/Sentinel/wiki/%E6%8E%A7%E5%88%B6%E5%8F%B0

Sentinel 控制台包含如下功能:

查看机器列表以及健康情况：收集 Sentinel 客户端发送的心跳包，用于判断机器是否在线。

监控 (单机和集群聚合)：通过 Sentinel 客户端暴露的监控 API，定期拉取并且聚合应用监控信息，

最终可以实现秒级的实时监控。

规则管理和推送：统一管理推送规则。

鉴权：生产环境中鉴权非常重要。这里每个开发者需要根据自己的实际情况进行定制。

1)jar包运行方式安装

Sentinel控制台下载地址：https://github.com/alibaba/Sentinel/releases

jar包下载后，直接启动即可，启动命令如下：

java -Dserver.port=8858 -Dcsp.sentinel.dashboard.server=localhost:8080 -

Dproject.name=sentinel-dashboard -jar sentinel-dashboard-1.7.2.jar

控制台访问：http://localhost:8858 效果如下：

登录的账号密码都是

sentinel。

2)Docker安装

访问http://192.168.211.137:8858/ 登录后，效果如下：

3.3 Sentinel案例

我们基于SpringCloud工程集成Sentinel，实现限流操作。为了节省时间，我们可以直接把写好的半成

品案例导入到IDEA中来，将资料\sentinel 中的sentinel工程导入进来，分别有一个生产者一个消费者

和一个微服务网关。

生产者： provider

消费者： consumer

微服务网关： gateway

docker pull bladex/sentinel-dashboard

docker run --name sentinel -d -p 8858:8858 -d bladex/sentinel-dashboard

3.3.1 基于Feign的服务降级

因为我们项目中服务之间调用使用的是feign，因此这里只讲解基于feign的服务降级实现。

1)引入依赖包：

在消费者中引入sentinel的包。

2)Sentinel配置

Sentinel 适配了 Feign 组件。如果想使用，除了引入 spring-cloud-starter-alibaba-sentinel 的依赖外

还需要开启Sentinel对feign的支持，并且配置控制台信息，我们可以在bootstrap.yml中配置，配置如

下：

3)配置服务降级方法

服务降级的方法就是原来springcloud中的服务降级配置，这里的配置完全是feign的配置，所以不详细

讲解了。

创建服务降级处理工厂对象 com.itheima.feign.fallback.GoodsFeignFallback ，代码如下：

<!-- spring c

loud alibaba sentinel 依赖 -->

<dependenc

<grou

d>com.alibaba.cloud</groupId>

<artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>

</depend

ency>

spring:

cloud:

#sentinel

sentinel:

transport:

port: 8719

dashboard: 192.168.211.137:8858

#Sentinel对feign的支持

feign:

sentinel:

enabled: true

在feign上添加fallbackFactory指向服务降级的工厂类对象，代码如下：

3.3.2 Sentinel控制台使用

我们接着启动生产者和消费者,并访问我们的控制台：http://192.168.211.137:8858/#/dashboard/ho

me，效果如下：

指定方法流量实时监控：

选择实时监控，再输入被访问的方法名字，就会出现指定方法访问的流量报表，如下图：

@Component

public class GoodsFeignFallback implements FallbackFactory<GoodsFeign> {

@Override

public GoodsFeign create(Throwable throwable) {

return new GoodsFeign() {

@Override

public String one(Integer max) {

return "服务降级";

}

};

}

@FeignClient(value = "goods",fallbackFactory = GoodsFeignFallback.class)

public interface GoodsFeign {

/***

* 获取一件商品

@GetMapping(value = "/goods/one/{max}")

String one(@PathVariable(value = "max")Integer max);

}

簇点链路

这块主要是单个节点中所有资源以及实时的调用数据，如下图：

3.3.2.1 流量控制

流控：

在上图中，针对每个资源都有3个操作按钮，其中流控主要用于做流量控制操作，其原理是监控应用流

量的 QPS 或并发线程数等指标，当达到指定的阈值时对流量进行控制，以避免被瞬时的流量高峰冲

垮，从而保障应用的高可用性。

讲解：

流控效果：

3.3.2.2 降级规则

RT：

异常比例：

resource：资源名，即限流规则的作用对象

count: 限流阈值

grade: 限流阈值类型（QPS 或并发线程数）

limitApp: 流控针对的调用来源，若为 default 则不区分调用来源

strategy: 调用关系限流策略

controlBehavior: 流量控制效果（直接拒绝、Warm Up、匀速排队）

快速失败：

当QPS超过任何规则的阈值后，新的请求就会立即拒绝，拒绝方式为抛出FlowException . 这种方

式适用于对

系统处理能力确切已知的情况下，比如通过压测确定了系统的准确水位时。

Warm Up：

当系统长期处理低水平的情况下，当流量突然增加时，直接把系统拉升到高水位可能瞬间把系统压

垮。通过"冷启动"，让通过的流量缓慢增加，在一定时间内逐渐增加到阈值的上限，给系统一个预热的时

间，避免冷系统被压垮。

排队等待：

匀速排队严格控制请求通过的时间间隔，也即是让请求以均匀的速度通过，对应的是漏桶算法。

异常数量：

3.3.2.3 热点规则

热点即经常访问的数据。很多时候我们希望统计某个热点数据中访问频次最高的 Top K 数据，并对其访

问进行限制。

参数说明：

Field

说明

默认值

resource

资源名，即限流规则的作用对象

limitApp

对应的黑名单/白名单，不同 origin 用 , 分隔，如

appA,appB

default，代表不区

分调用来源

strategy

限制模式， AUTHORITY_WHITE 为白名单模式，

AUTHORITY_BLACK 为黑名单模式，默认为白名单模式

AUTHORITY_WHITE

3.3.2.4 黑白名单

参数说明：

3.4 微服务网关Sentinel限流

在秒杀项目中，我们需要集成Sentinel进行限流操作，有些商品目前属于非热点商品，但也有可能存在

部分商品会在中途突然增加抢购热度成为热点，成为热点有可能会导致瞬间的流量膨胀，这时候我们可

以采用Sentinel做限流操作，以实现对后台微服务的保护。后台秒杀我们在微服务网关进行限流操作。

1)引入依赖

微服务网关集成Sentinel限流，需要引入如下依赖包：

2)配置控制台信息

在bootstrap.yml中配置endpoint以及控制台信息，配置如下：

现在所有的集成就成功了，我们启动微服务网关，看看Sentinel控制台也会显示微服务网关的节点信

息，并且可以在控制台配置限流策略。

<groupId>com.alibaba.cloud</groupId>

<artifactId>spring-cloud-alibaba-sentinel-gateway</artifactId>

</dependency>

<groupId>com.alibaba.cloud</groupId>

<artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>

</dependency>

spring:

cloud:

#sentinel

sentinel:

transport:

port: 8719

dashboard: 192.168.211.137:8858

#endpoint

management:

endpoints:

web:

exposure:

include: '*'

4 Lvs+Nginx集群配置

4.1 Lvs介绍

LVS（Linux Virtual Server）即Linux虚拟服务器，是由章文嵩博士主导的开源负载均衡项目，目前LVS

已经被集成到Linux内核模块中。该项目在Linux内核中实现了基于IP的数据请求负载均衡调度方案，其

体系结构如图1所示，终端互联网用户从外部访问公司的外部负载均衡服务器，终端用户的Web请求会

发送给LVS调度器，调度器根据自己预设的算法决定将该请求发送给后端的某台

Web服务器，比如，轮

询算法可以将外部的请求平均分发给后端的所有服务器，终端用户访问

LVS调度器虽然会被转发到后端

真实的服务器，但如果真实服务器连接的是相同的存储，提供的服务也是相同的服务，最终用户不管是

访问哪台真实服务器，得到的服务内容都是一样的，整个集群对用户而言都是透明的。

工作中采用Lvs集群模式：

4.2 Lvs工作解析模式介绍

NAT模式：即网络地址转换

访问步骤：

TUN负载均衡模式（IP隧道）：

在LVS（NAT）模式的集群环境中，由于所有的数据请求及响应的数据包都需要经过LVS调度器转发，如

果后端服务器的数量大于10台，则调度器就会成为整个集群环境的瓶颈。我们知道，数据请求包往往远

小于响应数据包的大小。因为响应数据包中包含有客户需要的具体数据，所以LVS（TUN）的思路就是

将请求与响应数据分离，让调度器仅处理数据请求，而让真实服务器响应数据包直接返回给客户端。

VS/TUN工作模式拓扑结构如图3所示。其中，IP隧道（IP tunning）是一种数据包封装技术，它可以将

原始数据包封装并添加新的包头（内容包括新的源地址及端口、目标地址及端口），从而实现将一个目

标为调度器的VIP地址的数据包封装，通过隧道转发给后端的真实服务器（Real Server），通过将客户

端发往调度器的原始数据包封装，并在其基础上添加新的数据包头（修改目标地址为调度器选择出来的

真实服务器的IP地址及对应端口），LVS（TUN）模式要求真实服务器可以直接与外部网络连接，真实

服务器在收到请求数据包后直接给客户端主机响应数据。

1.用户通过互联网DNS服务器解析到公司负载均衡设备上面的外网地址，相对于真实服务器而言，LVS外网

IP又称VIP（Virtual IP Address），用户通过访问VIP，即可连接后端的真实服务器（Real

Server），而这一切对用户而言都是透明的，用户以为自己访问的就是真实服务器，但他并不知道自己访

问的VIP仅仅是一个调度器，也不清楚后端的真实服务器到底在哪里、有多少真实服务器。

2.用户将请求发送至124.126.147.168，此时LVS将根据预设的算法选择后端的一台真实服务器

（192.168.0.1~192.168.0.3），将数据请求包转发给真实服务器，并且在转发之前LVS会修改数据包

中的目标地址以及目标端口，目标地址与目标端口将被修改为选出的真实服务器IP地址以及相应的端口。

3.真实的服务器将响应数据包返回给LVS调度器，调度器在得到响应的数据包后会将源地址和源端口修改为

VIP及调度器相应的端口，修改完成后，由调度器将响应数据包发送回终端用户，另外，由于LVS调度器有

一个连接Hash表，该表中会记录连接请求及转发信息，当同一个连接的下一个数据包发送给调度器时，从

该Hash表中可以直接找到之前的连接记录，并根据记录信息选出相同的真实服务器及端口信息。

DR负载均衡模式（推荐：直接路由模式）：

在LVS（TUN）模式下，由于需要在LVS调度器与真实服务器之间创建隧道连接，这同样会增加服务器

的负担。与LVS（TUN）类似，DR模式也叫直接路由模式，其体系结构如图4所示，该模式中LVS依然仅

承担数据的入站请求以及根据算法选出合理的真实服务器，最终由后端真实服务器负责将响应数据包发

送返回给客户端。与隧道模式不同的是，直接路由模式（DR模式）要求调度器与后端服务器必须在同

一个局域网内，VIP地址需要在调度器与后端所有的服务器间共享，因为最终的真实服务器给客户端回

应数据包时需要设置源IP为VIP地址，目标IP为客户端IP，这样客户端访问的是调度器的VIP地址，回应

的源地址也依然是该VIP地址（真实服务器上的VIP），客户端是感觉不到后端服务器存在的。由于多台

计算机都设置了同样一个VIP地址，所以在直接路由模式中要求调度器的VIP地址是对外可见的，客户端

需要将请求数据包发送到调度器主机，而所有的真实服务器的VIP地址必须配置在Non-ARP的网络设备

上，也就是该网络设备并不会向外广播自己的MAC及对应的IP地址，真实服务器的VIP对外界是不可见

的，但真实服务器却可以接受目标地址VIP的网络请求，并在回应数据包时将源地址设置为该VIP地址。

调度器根据算法在选出真实服务器后，在不修改数据报文的情况下，将数据帧的MAC地址修改为选出的

真实服务器的MAC地址，通过交换机将该数据帧发给真实服务器。整个过程中，真实服务器的VIP不需

要对外界可见。4.3 Lvs-DR配置

综合上面分析，我们可以得出结论，DR模式性能效率比较高，安全性很高，因此一般公司都推荐使用

DR模式。我们这里也配置

DR模式实现Lvs+Nginx集群。

3台机器：

VIP:

4.3.1 Vip配置

关闭网络配置管理器(每台机器都要做)

配置虚拟IP(VIP)

在 /etc/sysconfig/network-scripts 创建文件 ifcfg-ens33:1 ，内容如下：

重启网络服务：

我们可以看到在原来的网卡上面添加了一个虚拟IP 130：

同时需要对 192.168.211.137 、 192.168.211.138 构建虚拟机IP，但只是用于返回数据，而不能被用

户访问到，这时候需要操作 ifcfg-lo 。

192.168.211.136 192.168.211.141(Lvs)

192.168.

(

)

192.168.

(

)

192.168.211.130 192.168.211.140(VIP)

systemctl stop NetworkManager

systemctl disable NetworkManager

BOOTPROTO=static

DEVICE=ens33:1

ONBOOT=yes

IPADDR=192.168.211.130

NETMASK=255.255.255.0

service network restart

IPADDR=127.0.0.1，这里127.0.0.1属于本地回环地址，不属于任何一个有类别地址类。它代表设备的

本地虚拟接口，所以默认被看作是永远不会宕掉的接口。

NETMASK=255.255.255.255

192.168.211.137 ：

将 ifcfg-lo 拷贝一份 ifcfg-lo:1 ,并修改 ifcfg-lo:1 配置，内容如下：

刷新lo：

查看IP可以发现lo下多了130ip：

192.168.211.138 :

操作同上。

4.3.2 LVS集群管理工具安装

ipvsadm用于对lvs集群进行管理，需要手动安装。

安装命令：

版本查看：

效果如下：

4.3.3 ARP配置（地址解析协议）

ifup lo

yum install ipvsadm

ipvsadm -Ln

arp_ignore和arp_announce参数都和ARP协议相关，主要用于控制系统返回arp响应和发送arp请求时

的动作。这两个参数很重要，特别是在LVS的DR场景下，它们的配置直接影响到DR转发是否正常。

arp-ignore：arp_ignore参数的作用是控制系统在收到外部的arp请求时，是否要返回arp响应（0~8，

2-8用的很少）

arp-announce：ARP

通告行为：

配置文件： /etc/sysctl.conf ，将如下文件拷贝进去：

刷新配置：

添加路由（真实服务中添加【192.168.211.142/192.168.211.143】）：

添加了一个host地址，目的是用于接收数据报文，接收到了数据报文后会交给lo:1处理。（防止关机失

效，需要将上述命令添加到/etc/rc.local中）

添加完host后，可以查看一下： route -n ,效果如下：

此时如果无法识别route，需要安装相关工具 yum install net-tools 。

上述配置我们同样要在 192.168.211.138 中配置。

4.3.4 集群配置

ipvsadm命令讲解：

0，只要本机配置了IP，就能响应请求。

1，请求的目标地址到达对应的网络接口，才能响应对应请求

0，本机上任

何网络接口都向外通告（也就是任何请求都对用户进行响应），所有网卡都能接收通告

1，尽可能避

免

本

网

卡与

不

匹配

的

目标进行通告

2，只在本

网卡

进

行

通

告

（推

荐

）

net.ipv4.conf.all.arp_ignore = 1

net.ipv4.conf.default.arp_ignore = 1

net.ipv4.conf.lo.arp_ignore = 1

net.ipv4.conf.all.arp_announce = 2

net.ipv4.conf.default.arp_announce = 2

net.ipv4.conf.lo.arp_announce = 2

sysctl -p

route add -host 192.168.211.130 dev lo:1

算

法

说明

轮询算法，它将请求依次分配给不同的rs节点，也就是RS节点中均摊分配。这种算法简

单，但只适合于RS节点处理性能差不多的情况

wrr

加权轮训调度，它将依据不同RS的权值分配任务。权值较高的RS将优先获得任务，并且

分配到的连接数将比权值低的RS更多。相同权值的RS得到相同数目的连接数。

Wlc

加权最小连接数调度，假设各台RS的全职依次为Wi，当前tcp连接数依次为Ti，依次去

Ti/Wi为最小的RS作为下一个分配的RS

目的地址哈希调度（destination hashing）以目的地址为关键字查找一个静态hash表来

获得需要的RS

源地址哈希调度（source hashing）以源地址为关键字查找一个静态hash表来获得需要

的RS

最小连接数调度（least-connection）,IPVS表存储了所有活动的连接。LB会比较将连接请

求发送到当前连接最少的RS.

Lblc

基于地址的最小连接数调度（locality-based least-connection）：将来自同一个目的地

址的请求分配给同一台RS，此时这台服务器是尚未满负荷的。否则就将这个请求分配给

连接数最小的RS，并以它作为下一次分配的首先考虑。

添加集群TCP服务地址：（外部请求由该配置指定的

VIP处理）

参数说明：

负载均衡算法：

配置rs(2个)节点：

ipvsadm -A:用于创建集群

ipvsadm -E:用于修改集群

ipvsadm -D:用于删除集群

ipvsadm -C:用于清除集群数据

ipvsadm -R:用于重置集群配置规则

ipvsadm -S:用于保存修改的集群规则

ipvsadm -a:用于添加一个rs节点

ipvsadm -e:用于修改一个rs节点

ipvsadm -d:用于

删除一个rs节点

ipvsadm

-A -t 192.168.211.130:80 -s rr

-A：添加集群配置

-t：TCP请求地址(VIP)

-s：负载均衡算法

ipvsadm -a -t 192.168.211.130:80 -r 192.168.211.137:80 -g

ipvsadm -a -t 192.168.211.130:80 -r 192.168.211.138:80 -g

参数说明：

添加了节点后，我们通过ipvsadm -Ln查看，可以看到多了2个节点。

此时访问VIP： 192.168.211.130 的时候，发现一直是138 rs，原因是因为lvs这里有一个用户请求持久

化操作，会将用户请求的数据持久化，下次请求的时候，会从持久化数据中取出来，如果有持久化数

据，就按持久化数据中进行访问，没有则轮询，因此我们需要配置持久化时间。

此时我们来查看集群列表，会多了持久化时间：

设置tcp

效果如下：

访问 192.168.211.130 ，在输入 ipvsadm -Lnc ，效果如下：

-a：给集群添加一个节点

-t：指定VIP地址

-r：指定real server地址

-g：表示LVS的模式为dr模式

ipvsadm -E -t 192.168.211.130:80 -s rr -p 5

ipvsadm --set 2 2 2