seckill

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`);
});

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("插入数据库成功");
    });
})