SpringBootServerFrame/src/main/java/com/zero/barrageserver/dbwrite/base/BaseDBWriteQueue.java at master · zerochl/SpringBootServerFrame · GitHub

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
package com.zero.barrageserver.dbwrite.base;

import com.zero.barrageserver.common.base.BaseQueue;
import com.zero.barrageserver.common.manager.ExecutorManager;
import com.zero.barrageserver.dbwrite.DBWriteProducer;
import com.zero.barrageserver.dbwrite.entity.BaseDBWriteEntity;
import com.zero.barrageserver.dbwrite.iface.IDBWriteQueue;
import com.zero.barrageserver.common.base.BaseQueue;
import com.zero.barrageserver.dbwrite.entity.BaseDBWriteEntity;
import com.zero.barrageserver.dbwrite.iface.IDBWriteQueue;
import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.schedulers.Schedulers;
import lombok.extern.log4j.Log4j2;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.context.ServletContextAware;

import javax.servlet.ServletContext;

/**
 * Created by zero大神 on 2017/12/20.
 */
@Log4j2
public abstract class BaseDBWriteQueue extends BaseQueue<BaseDBWriteEntity> implements IDBWriteQueue {
    @Autowired
    private DBWriteProducer dbWriteProducer;
    @Override
    public void registerQueue() {
        dbWriteProducer.registerObserver(this);
    }

    @Override
    public void handleSync(BaseDBWriteEntity dbWriteEntity) {
        doExecute(dbWriteEntity);
    }

    @Override
    public void handleAsync(final BaseDBWriteEntity dbWriteEntity) {
//lamda格式，请注意
        Observable.create((ObservableOnSubscribe<Integer>) observableEmitter -> {
            log.info("on subscribe:" + ";thread:" + Thread.currentThread().getId());
            //执行写入操作
            doExecute(dbWriteEntity);
        })
        //此存储操作不需要考虑先后顺序，快速消费即可
        .subscribeOn(Schedulers.from(ExecutorManager.getDbWriteExecutor()))
        .subscribe();
    }
}