zhenghuang
/
bsdgy-server
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

门店号管理

Browse Source
dev_0531
LegnaYet 6 years ago
parent 9c3564620f
commit 1bca6e8a63
1 changed files with 95 additions and 95 deletions
Show Stats Download Patch File Download Diff File

190
src/main/java/com/kiisoo/ic/job/CustomerSyncJob.java
Unescape Escape View File

@ -1,95 +1,95 @@
package com.kiisoo.ic.job; //package com.kiisoo.ic.job;
//
import com.kiisoo.ic.customer.CustomerService; //import com.kiisoo.ic.customer.CustomerService;
import com.kiisoo.ic.store.entity.PrivilageCpUserStoreDO; //import com.kiisoo.ic.store.entity.PrivilageCpUserStoreDO;
import com.kiisoo.ic.store.mapper.PoiStoreStaffDOMapper; //import com.kiisoo.ic.store.mapper.PoiStoreStaffDOMapper;
import com.kiisoo.ic.store.mapper.PrivilageCpUserStoreDOMapper; //import com.kiisoo.ic.store.mapper.PrivilageCpUserStoreDOMapper;
import com.kiisoo.ic.wx.service.QWMailListManageService; //import com.kiisoo.ic.wx.service.QWMailListManageService;
import com.lmax.disruptor.*; //import com.lmax.disruptor.*;
import lombok.extern.slf4j.Slf4j; //import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; //import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service; //import org.springframework.stereotype.Service;
//
import java.util.List; //import java.util.List;
import java.util.concurrent.*; //import java.util.concurrent.*;
//
@Service //@Service
@Slf4j //@Slf4j
public class CustomerSyncJob { //public class CustomerSyncJob {
//
@Autowired // @Autowired
private PrivilageCpUserStoreDOMapper privilageCpUserStoreDOMapper; // private PrivilageCpUserStoreDOMapper privilageCpUserStoreDOMapper;
//
@Autowired // @Autowired
private QWMailListManageService qwMailListManageService; // private QWMailListManageService qwMailListManageService;
//
@Autowired // @Autowired
private PoiStoreStaffDOMapper poiStoreStaffDOMapper; // private PoiStoreStaffDOMapper poiStoreStaffDOMapper;
//
@Autowired // @Autowired
private CustomerService customerService; // private CustomerService customerService;
//
public void syncCustomer() throws ExecutionException, InterruptedException { // public void syncCustomer() throws ExecutionException, InterruptedException {
long beginTime = System.currentTimeMillis(); // long beginTime = System.currentTimeMillis();
//
int BUFFER_SIZE = 1024 * 1024; // int BUFFER_SIZE = 1024 * 1024;
int THREAD_NUMBERS = 10; // int THREAD_NUMBERS = 10;
/* // /*
* createSingleProducer创建一个单生产者的RingBuffer， // * createSingleProducer创建一个单生产者的RingBuffer，
* 第一个参数叫EventFactory，从名字上理解就是“事件工厂”，其实它的职责就是产生数据填充RingBuffer的区块。 // * 第一个参数叫EventFactory，从名字上理解就是“事件工厂”，其实它的职责就是产生数据填充RingBuffer的区块。
* 第二个参数是RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率 // * 第二个参数是RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率
* 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略 // * 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略
*/ // */
final RingBuffer<DTOMessage> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<DTOMessage>() { // final RingBuffer<DTOMessage> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<DTOMessage>() {
@Override // @Override
public DTOMessage newInstance() { // public DTOMessage newInstance() {
return new DTOMessage(); // return new DTOMessage();
} // }
}, BUFFER_SIZE, new YieldingWaitStrategy()); // }, BUFFER_SIZE, new YieldingWaitStrategy());
//创建线程池 // //创建线程池
ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS); // ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);
//创建SequenceBarrier // //创建SequenceBarrier
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier(); // SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
//
//创建消息处理器 // //创建消息处理器
BatchEventProcessor<DTOMessage> transProcessor = new BatchEventProcessor<DTOMessage>( // BatchEventProcessor<DTOMessage> transProcessor = new BatchEventProcessor<DTOMessage>(
ringBuffer, sequenceBarrier, new DTOMessageHandler()); // ringBuffer, sequenceBarrier, new DTOMessageHandler());
//
//这一部的目的是让RingBuffer根据消费者的状态 如果只有一个消费者的情况可以省略 // //这一部的目的是让RingBuffer根据消费者的状态 如果只有一个消费者的情况可以省略
ringBuffer.addGatingSequences(transProcessor.getSequence()); // ringBuffer.addGatingSequences(transProcessor.getSequence());
//
//把消息处理器提交到线程池 // //把消息处理器提交到线程池
executors.submit(transProcessor); // executors.submit(transProcessor);
//如果存大多个消费者 那重复执行上面3行代码 把TradeTransactionInDBHandler换成其它消费者类 // //如果存大多个消费者 那重复执行上面3行代码 把TradeTransactionInDBHandler换成其它消费者类
//
Future<?> future = executors.submit(new Callable<Void>() { // Future<?> future = executors.submit(new Callable<Void>() {
@Override // @Override
public Void call() throws Exception { // public Void call() throws Exception {
List<PrivilageCpUserStoreDO> privilageCpUserStoreDOS = privilageCpUserStoreDOMapper.selectList(null); // List<PrivilageCpUserStoreDO> privilageCpUserStoreDOS = privilageCpUserStoreDOMapper.selectList(null);
long seq; // long seq;
for (PrivilageCpUserStoreDO privilageCpUserStoreDO : privilageCpUserStoreDOS) { // for (PrivilageCpUserStoreDO privilageCpUserStoreDO : privilageCpUserStoreDOS) {
String cpUserId = privilageCpUserStoreDOMapper.selectCpUserIdByStoreId(privilageCpUserStoreDO.getStoreId()); // String cpUserId = privilageCpUserStoreDOMapper.selectCpUserIdByStoreId(privilageCpUserStoreDO.getStoreId());
//
seq = ringBuffer.next();//占个坑 --ringBuffer一个可用区块 // seq = ringBuffer.next();//占个坑 --ringBuffer一个可用区块
//
//给这个区块放入 数据 如果此处不理解，想想RingBuffer的结构图 // //给这个区块放入 数据 如果此处不理解，想想RingBuffer的结构图
ringBuffer.get(seq).setCpUserId(cpUserId); // ringBuffer.get(seq).setCpUserId(cpUserId);
ringBuffer.get(seq).setPrivilageCpUserStoreDO(privilageCpUserStoreDO); // ringBuffer.get(seq).setStoreId(privilageCpUserStoreDO);
ringBuffer.get(seq).setQwMailListManageService(qwMailListManageService); // ringBuffer.get(seq).setQwMailListManageService(qwMailListManageService);
ringBuffer.get(seq).setCustomerService(customerService); // ringBuffer.get(seq).setCustomerService(customerService);
ringBuffer.get(seq).setPoiStoreStaffDOMapper(poiStoreStaffDOMapper); // ringBuffer.get(seq).setPoiStoreStaffDOMapper(poiStoreStaffDOMapper);
// ringBuffer.get(seq).setCustomers(customers); //// ringBuffer.get(seq).setCustomers(customers);
ringBuffer.publish(seq);//发布这个区块的数据使handler(consumer)可见 // ringBuffer.publish(seq);//发布这个区块的数据使handler(consumer)可见
//
} // }
return null; // return null;
} // }
}); // });
future.get();//等待生产者结束 // future.get();//等待生产者结束
Thread.sleep(1000);//等上1秒，等消费都处理完成 // Thread.sleep(1000);//等上1秒，等消费都处理完成
transProcessor.halt();//通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!） // transProcessor.halt();//通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!）
executors.shutdown();//终止线程 // executors.shutdown();//终止线程
//
System.out.println("总耗时:" + (System.currentTimeMillis() - beginTime)); // System.out.println("总耗时:" + (System.currentTimeMillis() - beginTime));
} // }
} //}

Write Preview
Loading…
Cancel
Save