一：环境

1.一台阿里云服务器
2.云服务器需要安装redis、kafka、mysql、zookeeper环境
3.一台本地电脑需要安装postman
注：kafka与zookeeper的安装和配置请参考：https://blog.csdn.net/wushichao0325/article/details/84993081 或者使用docker直接拉去镜像安装。

二：工具说明

1.postman：用于使用其中的（Run）实现并发测试
2.redis:用于存放商品信息，这里只创建一个counter字段，使用redis的命令set设置个初始值。100，即商品总量。模拟数据为：counter=100
3.kafka：结合zookeeper使用，将秒杀成功的结果生成一个producer，待consumer去消费及同步数据到mysql。模拟数据为：创建一个topic名为PRODUCT_NUMBER。
4.mysql：存放最终的秒杀结果。模拟数据为：创建一个seckill数据库，再创建一个seckill表，表字段为id自增和date存放秒杀时间和uid存放用户id。

三：代码实现

1.代码目录

2.node依赖包

3.seckillService.js
接收秒杀请求，并将秒杀结果及用户信息发送个kafka的producer。

var express=require("express"),
    redis=require("redis"),
    kafka=require('kafka-node'),
    Producer=kafka.Producer,
    kafkaClient=new kafka.Client(),
    producer=new Producer(kafkaClient);
    count=0;
    app=express();
    config=require("./config/redis.json");
var bodyParser=require('body-parser');
app.use(express.json());
app.use(bodyParser.json({limit: '1mb'}));  //body-parser 解析json格式数据
app.use(bodyParser.urlencoded({            //此项必须在 bodyParser.json 下面,为参数编码
    extended: true
}));
app.all("*", function (req, res, next) {
    res.setHeader("Access-Control-Allow-Origin", "*");
    res.setHeader('Access-Control-Allow-Headers', 'Content-Type, Content-Length, Authorization, Accept, X-Requested-With , yourHeaderFeild');
    res.setHeader("Access-Control-Allow-Methods","PUT,POST,GET,DELETE,OPTIONS");
    res.setHeader("X-Powered-By",' 3.2.1');
    next();
});
app.post('/seckill',function(req,res){
    console.log('count='+count++);
    let fn=function(optionalClient){
        let client;
        if(optionalClient=='undefined'||optionalClient==null){
            client=redis.createClient(config);
        }else{
            client=optionalClient;
        }
        client.on('error',function(er){
            console.error(er.stack);
            client.end(true);
        });
        client.watch("counter");//监听counter字段
        client.get("counter",function(err,reply){
            if(parseInt(reply)>0){
                let multi=client.multi();
                multi.decr('counter');//更新redis的counter数量减一。
                multi.exec(function(err,replies){
                    if(replies==null){//counter字段正在操作中，等待counter被其他释放
                        console.log("counter被使用");
                        fn(client);
                    }else{
                        var args = {
                            openid: 'b05NZ2Y1WjbE9fRV9MZTBWWQ==',
                            seckillTime: '2018-12-12 00:00:01',
                        }
                        let payload=[{
                            topic:'PROUDCT_NUMBER',
                            messages:[JSON.stringify(args)],
                            key:"seckill",
                            partition:0
                        }];
                        console.log("payload:",payload);
                        producer.send(payload,function(err,data){
                            console.log(data);
                        });
                        res.send(replies);
                        client.end(true);
                    }
                })

            }else{
                console.log("已经卖光了");
                res.send("已经卖光了");
                client.end(true);
            }
        });
    }
    fn(null);
});
app.listen(8888, "0.0.0.0",function () {
    console.log(`Please link connectorServer http://localhost:8888`);
});

4.seckill_kafka_consumer.js
消费者消费生产者的数据，并将数据同步到mysql中。

var kafka=require('kafka-node'),
    Consumer=kafka.Consumer,
    client=new kafka.Client(),
    consumer=new Consumer(
        client,
        [
            {topic:'CAR_NUMBER',partition:0}
        ],
        {
            autoCommit:false,
            fetchMaxWaitMs: 1000,
            fetchMaxBytes: 1024 * 1024,
            fromOffset: true
        }
    );
var mysql=require('mysql');
var connection=mysql.createConnection({
    "host":"**.***.***.**",
    "user":"root",
    "password":"password",
    "database":"seckill"
});
connection.connect();
consumer.on('message',function(message){
    console.log("得到的生产者的数据为：",message);
    let value=JSON.parse(message.value);
    connection.query('insert into seckill set ?',{date:new Date(),uid:value.openid},function(error,results,fields){
        if(error){
            console.error(error);
        }
        console.log("插入数据库成功");
    });
})

四：启动项目

1.启动云服务器的redis
2.启动云服务器的zookeeper
3.启动云服务器的kafka
4.将项目部署到云服务器中。
5.进入项目根目录，启动seckillService.js和seckill_kafka_consumer.js，我是使用pm2去管理这两个服务。

五：测试

打开postman如图：

先将这个链接保存一个文件，然后点击Runner按钮。
如图：

可以使用post请求：http://seckill.ykplay.com/seckill 测试。
服务器打印信息如下：
seckillService.js

seckill_kafka_consumer.js

有这些信息证明秒杀功能成功实现了。