zoukankan      html  css  js  c++  java
  • 异步远程调用线程池队列

    import com.google.common.util.concurrent.ThreadFactoryBuilder;
    import lombok.extern.slf4j.Slf4j;
    import java.util.concurrent.*;

    1、实现类

    @Slf4j
    public class MucRpcThreadPoolExecutor {
    /**
    * 任务队列
    * 设置有界队列长度为1024
    */
    private static BlockingQueue<Runnable> queue = new ArrayBlockingQueue<Runnable>(1024);

    private static ExecutorService executorService = new ThreadPoolExecutor(
    4, 16, 60, TimeUnit.SECONDS,
    queue,
    new ThreadFactoryBuilder().setNameFormat("muc-rpc-thd-%d").build(),
    new RpcExecutionHandler());

    public static void asyncRpcSubmit(AbstractRpcRunner runner){
    executorService.submit(runner);
    }


    private static class RpcExecutionHandler implements RejectedExecutionHandler{
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
    String name = Thread.currentThread().getName();
    if (r instanceof AbstractRpcRunner){
    if (((AbstractRpcRunner) r).getBaseRpcBo() != null)
    name = ((AbstractRpcRunner) r).getBaseRpcBo().getReqId();
    }
    log.warn("muc-rpc-thread-pool线程队列已满,新任务被丢弃:{}", name);
    }
    }
    }
    2、线程抽象类
    //BaseRpcBo 请求实体类
    public  abstract class AbstractRpcRunner <T extends BaseRpcBo> implements Runnable {
    protected T baseRpcBo;


    public T getBaseRpcBo() {
    return baseRpcBo;
    }

    public void setBaseRpcBo(T baseRpcBo) {
    this.baseRpcBo = baseRpcBo;
    }
    }

    3、实体类
    @Data
    public class BaseRpcBo implements Serializable {
    /**
    * 请求id
    */
    private String reqId;

    /**
    * 远程调用地址
    */
    @JSONField(serialize = false)
    private String rpcUrl;

    /**
    * 请求头
    */
    @JSONField(serialize = false)
    private Map<String, String> headers;
    }

    4、调用方法
    public void asyncGrowthUp(String uid) {
    GrowthUpRpcBo rpcBo = new GrowthUpRpcBo();
    rpcBo.setStamp(DateFormatUtils.format(new Date(), "yyyyMMddHHmmss"));
    rpcBo.setUid(uid);
    rpcBo.setReqId(UUIDUtil.randomId());
    rpcBo.setRpcUrl(mjHttpUtils.getAbsUrl(MjConstant.API_GROWTH_UP_SYNC));
    Map<String,String> header = new HashMap<>(1);
    header.put("Content-Type", "application/json;charset=UTF-8");
    rpcBo.setHeaders(header);
    MucRpcThreadPoolExecutor.asyncRpcSubmit(new PostJsonRpcRunner(mjHttpUtils, rpcBo));
    log.info("growth up notify to iot finish, uid:{}, reqId:{}", uid, rpcBo.getReqId());
    }
    5、GrowthUpRpcBo
    @Data
    public class GrowthUpRpcBo extends BaseRpcBo{
    /**
    * 请求时间戳
    * 格式为 yyyyMMddHHmmss
    */
    String stamp;
    /**
    * 用户id
    */
    String uid;
    }




  • 相关阅读:
    c++类中的常量
    什么是租赁
    期末结转销售成本结算利润
    合并报表抵消分录借:营业成本 贷:存货怎么理解(自己分析)
    电脑上装安卓系统
    下载apps from google play on windows 10
    virtualbox找不到ubuntu64选项
    如何区分其他综合收益和当期损益
    哪些项目进其他综合收益
    境外经营财务报表的折算方法
  • 原文地址:https://www.cnblogs.com/fuqiang-terry/p/12529781.html
Copyright © 2011-2022 走看看