手写 kafka 异步回调

玩过 kafka 的小伙伴相信对 kafka 生产者和消费者的异步回调记忆犹新，不由的赞叹这种设计模式真的超赞，心想如果我也能写出这样的异步回调那该多好，今天他来了！！！

Callback

模仿 kafka 的这个接口用于接收回调信息

package demo;import java.util.Map;public interface Callback {void onCompletion(Map<String, String> offset, Exception exception);
}

ReomteServer

这个就相当于 kafka 实际发消息的类，将消息异步发送(也可以同步)给服务器后返回元数据信息。首先分析一下既要异步也要支持同步那就需要使用 Future 类了，利用它的 get() 方法实现同步即可!!!

package demo;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;public class RemoteServer<T> {private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");public Future<Map<String, String>> send(T message) {return send(message, null);}public Future<Map<String, String>> send(T message, Callback callback) {FutureTask<Map<String, String>> futureTask = new FutureTask<>(() -> {if (callback != null) {if ("error".equals(message)) {callback.onCompletion(null, new RuntimeException(sdf.format(new Date())+"因为你发了error，所以报错了"));} else {callback.onCompletion(getMap(message), null);}}return getMap(message);});new Thread(futureTask).start();return futureTask;}public Map<String, String> getMap(T message) {HashMap<String, String> map = new HashMap<>();map.put("message", message.toString());map.put("offset", message.hashCode() + "");map.put("time", sdf.format(new Date()));return map;}
}

这样一个简单的异步回调就完成啦。

LocalServer

本地测试

正常消息异步发送+回调

package demo;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ExecutionException;public class LocalServer {public static void main(String[] args) throws ExecutionException, InterruptedException {String message = "message";RemoteServer<String> remoteServer = new RemoteServer<>();for (int i = 0; i < 10; i++) {remoteServer.send(message, new Callback() {@Overridepublic void onCompletion(Map<String, String> offset, Exception exception) {if (null == exception) {System.out.println(offset);} else {exception.printStackTrace();}}});System.out.println(new Date() + "发过去了");}}
}

测试结果

正常消息同步发送

package demo;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ExecutionException;public class LocalServer {public static void main(String[] args) throws ExecutionException, InterruptedException {String message = "error";RemoteServer<String> remoteServer = new RemoteServer<>();for (int i = 0; i < 10; i++) {System.out.println(remoteServer.send(message).get());System.out.println(new Date() + "发过去了");}}
}

错误消息异步发送+回调

package demo;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ExecutionException;public class LocalServer {public static void main(String[] args) throws ExecutionException, InterruptedException {String message = "error";RemoteServer<String> remoteServer = new RemoteServer<>();for (int i = 0; i < 10; i++) {remoteServer.send(message, new Callback() {@Overridepublic void onCompletion(Map<String, String> offset, Exception exception) {if (null == exception) {System.out.println(offset);} else {exception.printStackTrace();}}});System.out.println(new Date() + "发过去了");}}
}

手写 kafka 异步回调相关推荐

阿里P8架构大神分享纯手写“kafka文档”看完直呼太牛！
什么是 Kafka Kafka 是由 Linkedin 公司开发的,它是一个分布式的,支持多分区.多副本,基于 Zookeeper 的分布式消息流平台,它同时也是一款开源的基于发布订阅模式的消息引擎系 ...
python3多进程写Kafka异步线程调用接口
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Author: 风过无言花易落 # @Date : 2022/02/14 22:30 # @Desc ...
promise 浏览器实现的源码_【大前端01-01】函数式编程与JS异步编程、手写Promise...
[简答题]一.谈谈你是如何理解JS异步编程的,EventLoop.消息队列都是做什么的,什么是宏任务.什么是微任务? 如何理解JS异步编程众所周知JavaScript语言执行环境是"单线程 ...
前端为什么有的接口明明是成功回调却执行了.catch失败回调_前端进阶高薪必看-手写源码篇（高频技术点）...
前言此系列作为笔者之前发过的前端高频面试整理的补充会比较偏向中高前端面试问题当然大家都是从新手一路走过来的感兴趣的朋友们都可以看哈初衷我相信不少同学面试的时候最怕的一个环节就是手写代码大 ...
【学习笔记】Part1·JavaScript·深度剖析-函数式编程与 JS 异步编程、手写 Promise（二、JavaScript 异步编程）
[学习笔记]Part1·JavaScript·深度剖析-函数式编程与 JS 异步编程.手写 Promise(课前准备) [学习笔记]Part1·JavaScript·深度剖析-函数式编程与 JS 异步 ...
mysql异步查询 java_java 手写并发框架（一）异步查询转同步的 7 种实现方式
序言本节将学习一下如何实现异步查询转同步的方式,共计介绍了 7 种常见的实现方式. 思维导图如下: 异步转同步业务需求有些接口查询反馈结果是异步返回的,无法立刻获取查询结果. 比如业务开发中我们 ...
并发查询_java 手写并发框架（一）异步查询转同步的7种实现方式
序言本节将学习一下如何实现异步查询转同步的方式,共计介绍了 7 种常见的实现方式. 思维导图如下: 思维导图异步转同步业务需求有些接口查询反馈结果是异步返回的,无法立刻获取查询结果. 比如业务 ...
react 日期怎么格式化_手写React的Fiber架构，深入理解其原理
熟悉React的朋友都知道,React支持jsx语法,我们可以直接将HTML代码写到JS中间,然后渲染到页面上,我们写的HTML如果有更新的话,React还有虚拟DOM的对比,只更新变化的部分,而不重 ...
手写webpack系列一：了解认识loader-utils
Created By JishuBao on 2019-03-29 12:38:22 Recently revised in 2019-04-01 12:38:22 欢迎大家来到技术宝的掘金世界, ...

手写 kafka 异步回调

Callback

ReomteServer

LocalServer

手写 kafka 异步回调相关推荐

最新文章

热门文章