一直想写一个秒杀系统的设计，网上有烂大街的教程，虽然有的写得很不错，但是还是自己设计的来的清楚，基本思路就是拦截+负载+缓存+异步,流程如下（文字改天补上）：

Topic 或者Subject

每一个生产者都需要向队列中生产消息，不同的生产者生产消息需要有所区别，供对应的消费者消费消息，这个是队列名称之为 Topic或者Subject

MessageLog，ConsumerLog

队列的消息需要一个存储的介质，Kafka的对应的存储为文件存储，生产者生产的消息存储在MessageLog, 然后根据不同的消费和路由规则路由，投递到对应的服务器上面，产生对应的ConsumerLog.

Partition

当投递的消息比较多的时候，就需要对ConsumerLog进行分片，分到不同的服务器上面，这个分片称之为partition,对于Kafka来说，一个Consumer一般和Partition成倍数关系，一个Consumer可以消费一个或者多个Partition.

Broken

Broken可以理解为消费者的服务器。

顺序IO

很多成熟的MQ的消息的存储都采用的磁盘的存储模式，可能有人会认为为什么不采用内存？内存的效率不应该更快吗？我开始也有这个疑问，后来才知道顺序IO的时候，才知道不适用内存的原因：

三次握手—连接

在通信的过程中，Client与Server建立TCP连接需要三次握手，为什么需要三次握手呢？又是怎么握手的过程？

TCP连接是可靠的通信方式，必须要保证两端都同时有效，且线路通畅。

如同两个人通话，但并不确定对方能不能听到，经历几次才能确保通信方式可靠呢？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

A: 请求连接，收到请回复密码1
B：可以连接，密码是1。你听到了吗？听到回复密码2
A：我听到了，密码2。

A就绪，B就绪。线路通畅,可以通话
```    

通过上面的分析，可以明确的知道为什么`TCP`需要三次握手，因为少了任何一次，都不能确保当前的网络是通畅的。 

Q:**使用两次握手会有什么问题？**   

A:  如果使用两次握手协议，B不知道A是否能够收到自己的消息，如果B此时进行补偿，就有可能多次发送。如果B此时忽略，A可能真的没有收到消息。
如果B的消息在传输的过程中丢失了，A也将不知道B有没有真正的准备好。



实际具体过程如下：   

 ![TCP](/img/assets/66/01.jpg)    


 1. `Server`主动开启，进入监听状态
 2. `Client`连接， 发送`syn`包到服务器，`Client`进入`SYN_SENT`状态
 3. 服务器收到syn包，也发送一个`sync`，同时也要确认`Client`的消息，所以发送`sync+ack`包，此时服务器进入SYN_RECV状态
 4. `Client`收到`Server`的`sync+ack`消息，发送一个`ack`消息服务器，进入连接状态
 5. 服务器收到`ack`消息进入连接就绪的状态
 

Q:**如果已经建立了连接，但是`Client`突然出现故障了怎么办**

A: 显然`Server`不可能一直等下去，不然会浪费很多资源。在`Server`有一个计时器，没当成功收到一次消息重置计时器，如果持续一段时间没有收到消息，服务器就会发送一个探测报文段， 每隔一段时间（75s）发送一次。 连续发送10个探测报文仍然没反应，服务器就认为`Client`出了故障，接着就关闭连接，释放资源。
 
----------------------------------------------------------------


#### 四次挥手--断开

 如果还是用通话的例子来举例的话，为了确保电话两端不至于异常挂断,应该怎么做？   

``` cte 
A：我要断开了
B: 好，等我一会，我先断开
B：我已经断开
A: 我已断开

A断开，B断开

实际具体过程如下：

1. Client进程发出断开消息，进入FIN-WAIT-1状态
2. Server收到断开消息，返回Ack报文。Server开始释放连接，Server进入CLOSE-WAIT
3. Client收到Ack回来的消息，进入FIN-WAIT-2状态
4. Server释放资源完成后，再次通知Client可以关闭连接，Server进入Last-Ack状态
5. Client收到服务器释放完成的消息后，经过2MSL（最长报文段寿命）的时间后，当客户端撤销相应的TCB后，才进入CLOSED状态。最终关闭完成后通知Server
6. Server收到最后Ack消息后，直接进入Closed状态。

分布式锁

在分布式环境中，为了保证业务数据的正常访问，防止出现重复请求的问题，会使用分布式锁来阻拦后续请求。具体伪代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

    public void doSomething(String userId){
        User user=getUser(userId);
        if(user==null){
            user.setUserName("xxxxx");
            user.setUserId(userId);
            insert(user);
            return;
        }
        update(user);
    }

```     

上面的代码很简单，查询db中有没有对应的user数据，如果有的话，执行更新操作，如果没有则插入。    

我们知道，上面的代码是线程不安全的，在多线程的环境中，就会出现问题。为了能够保证数据的正确性，在单机环境下，我们可以使用`synchronized`的方法，来保证线程安全，具体修改：  


``` Java   
    public synchronized void doSomething(String userId){
        User user=getUser(userId);
        if(user==null){
            user.setUserName("xxxxx");
            user.setUserId(userId);
            insert(user);
            return;
        }
        update(user);
    }

```     

在单机器的环境下，能够解决线程安全的问题，那在分布式环境下呢？ 这个时候需要用到`分布式锁`.   


分布式锁需要借助其他组件来实现，常用的有`redis`和`zookeeper`。下面我们就用redis的实现，来说明下问题，分布式锁具体的实现方法如下：  

``` Java 
    public  void doSomething(String userId){
        String lock=RedisUtils.get("xxxx"+userId);
        if(StringUtils.isNotEmpty(lock)){//说明当前userId已经被锁定
            return;
        }
        RedisUtils.set("xxxx"+userId,userId,1000);//锁定10s
        User user=getUser(userId);
        if(user==null){
            insert(user);
            RedisUtils.delete("xxxx"+userId);
            return;
        }
        update(user);
        RedisUtils.delete("xxxx"+userId);
        
    }

上面的代码解决了在分布式环境中的并发的问题。但同样需要考虑一个问题，如果insert操作和update操作异常了，分布式锁不会释放，后续的请求还会被拦截。

所以我们再优化，增加对异常的捕获。

    public  void doSomething(String userId){
        try {
                String lock=RedisUtils.get("xxxx"+userId);
                if(StringUtils.isNotEmpty(lock)){//说明当前userId已经被锁定
                    return;
                }
                RedisUtils.set("xxxx"+userId,userId,1000);//锁定1s
                User user=getUser(userId);
                if(user==null){
                    insert(user);
                    return;
                }
                update(user);
        }
        catch(Exception ex){

        }
        finally{
            RedisUtils.delete("xxxx"+userId);
        }
    }

现在即使是程序异常了，锁会自动释放。但redis的get和set也会存在并发问题，我们再继续优化，使用redis中的setnx方法。

    public  void doSomething(String userId){
        try {
                boolean lock=RedisUtils.setnx("xxxx"+userId,userId,1000);//锁定1s
                if(!lock){//说明当前userId已经被锁定
                    return;
                }
                User user=getUser(userId);
                if(user==null){
                    insert(user);
                    return;
                }
                update(user);
        }
        catch(Exception ex){

        }
        finally{
            RedisUtils.delete("xxxx"+userId);
        }
    }

上面的代码好像没有什么问题了，但也存在很大的隐患。
我们分析下，假设第一个请求过来，执行锁定成功，程序开始运行，但是insert和update操作阻塞了1s，第二个请求过来，锁的缓存已经过期，第二个执行锁定成功，这个时候第一个请求完成了锁被释放，第二个请求的锁就被第一次请求释放了，第三次的请求就会造成线程不安全问题。

怎么再去优化呢？问题主要是出现在第一次请求误删锁的问题，所以我们在移除锁的时候要判断能否移除。