（RPC框架基本结构）
（RPC调用流程）
（Tars Java初始化过程）
// 先初始化基本Tars配置
CommunicatorConfig cfg = new CommunicatorConfig();
// 通过上述的CommunicatorConfig配置生成一个Communicator对象。
Communicator communicator = CommunicatorFactory.getInstance().getCommunicator(cfg);
// 指定Tars远程服务的服务对象名、IP和端口生成一个远程服务代理对象。
HelloPrx proxy = communicator.stringToProxy(HelloPrx.class, "TestApp.HelloServer.HelloObj@tcp -h 127.0.0.1 -p 18601 -t 60000");
//同步调用，阻塞直到远程服务对象的方法返回结果
String ret = proxy.hello(3000, "Hello World");
System.out.println(ret);
//异步调用，不关注异步调用最终的情况
proxy.async_hello(null, 3000, "Hello World");
//异步调用,注册一个实现TarsAbstractCallback接口的回执处理对象，该实现类分别处理调用成功，调用超时和调用异常的情况。
proxy.async_hello(new HelloPrxCallback() {
@Override
public void callback_expired() { //超时事件处理
}
@Override
public void callback_exception(Throwable ex) { //异常事件处理
}
@Override
public void callback_hello(String ret) { //调用成功事件处理
Main.logger.info("invoke async method successfully {}", ret);
}
}, 1000, "Hello World");
在上述例子中，演示了常见的两种调用方式，分别为同步调用和异步调用。其中异步调用，如果调用方想捕捉异步调用的最终结果，可以注册一个实现TarsAbstractCallback接口的实现类，对tars调用的异常，超时和成功事件进行处理。
public final class ObjectProxy<T> implements ServantProxy, InvocationHandler {
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
String methodName = method.getName();
Class<?>[] parameterTypes = method.getParameterTypes();
InvokeContext context = this.protocolInvoker.createContext(proxy, method, args);
try {
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return this.toString();
} else if
//***** 省略代码 *****
} else {
// 在负载均衡器选取一个远程调用类，进行应用层协议的封装，最后调用TCP传输层进行发送。
Invoker invoker = this.loadBalancer.select(context);
return invoker.invoke(context);
}
} catch (Throwable var8) {
// ***** 省略代码 *****
}
}
}
当然生成上述远程服务代理类，涉及到辅助类，Tars Java采用ServantProxyFactory来生成上述的ObjectProxy，并存储ObjectProxy对象到Map结构，便于调用方二次使用时直接复用已存在的远程服务代理对象。
具体相关逻辑如源码所示，ObjectProxyFactory是生成ObjectProxy的辅助工厂类，和ServantProxyFactory不同，其本身不缓存生成的代理对象。
class ServantProxyFactory {
private final ConcurrentHashMap<String, Object> cache = new ConcurrentHashMap();
// ***** 省略代码 *****
public <T> Object getServantProxy(Class<T> clazz, String objName, ServantProxyConfig servantProxyConfig, LoadBalance loadBalance, ProtocolInvoker<T> protocolInvoker) {
Object proxy = this.cache.get(objName);
if (proxy == null) {
this.lock.lock(); // 加锁，保证只生成一个远程服务代理对象。
try {
proxy = this.cache.get(objName);
if (proxy == null) {
// 创建实现JDK的java.lang.reflect.InvocationHandler接口的对象
ObjectProxy<T> objectProxy = this.communicator.getObjectProxyFactory().getObjectProxy(clazz, objName, servantProxyConfig, loadBalance, protocolInvoker);
// 使用JDK的java.lang.reflect.Proxy来生成实际的代理对象
this.cache.putIfAbsent(objName, this.createProxy(clazz, objectProxy));
proxy = this.cache.get(objName);
}
} finally {
this.lock.unlock();
}
}
return proxy;
}
/** 使用JDK自带的Proxy.newProxyInstance生成代理对象 */
private <T> Object createProxy(Class<T> clazz, ObjectProxy<T> objectProxy) {
return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{clazz, ServantProxy.class}, objectProxy);
}
// ***** 省略代码 *****
}
从以上的源码中，可以看到createProxy使用了JDK的Proxy.newProxyInstance方法来生成远程服务代理对象。
（客户端按特定路由规则调用远程服务）
public interface LoadBalance<T> {
/** 根据负载均衡策略，挑选invoker */
Invoker<T> select(InvokeContext invokeContext) throws NoInvokerException;
/** 通知invoker列表的更新 */
void refresh(Collection<Invoker<T>> invokers);
}
public final class Reactor extends Thread {
protected volatile Selector selector = null;
private Acceptor acceptor = null;
//***** 省略代码 *****
public void run() {
try {
while (!Thread.interrupted()) {
// 阻塞直到有网络事件发生。
selector.select();
//***** 省略代码 *****
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();
if (!key.isValid()) continue;
try {
//***** 省略代码 *****
// 分发传输层协议TCP或UDP网络事件
dispatchEvent(key);
//***** 省略代码 *****
}
}
//***** 省略代码 *****
}
//***** 省略代码 *****
private void dispatchEvent(final SelectionKey key) throws IOException {
if (key.isConnectable()) {
acceptor.handleConnectEvent(key);
} else if (key.isAcceptable()) {
acceptor.handleAcceptEvent(key);
} else if (key.isReadable()) {
acceptor.handleReadEvent(key);
} else if (key.isValid() && key.isWritable()) {
acceptor.handleWriteEvent(key);
}
}
}
网络处理采用Reactor事件驱动模式，Tars定义一个Reactor对象对应一个Selector对象，针对每个远程服务（整体服务集群，非单个节点程序）默认创建2个Reactor对象进行处理，通过修改com.qq.tars.net.client.selectorPoolSize这个JVM启动参数值来决定一个远程服务具体创建几个Reactor对象。
（Tars-Java的网络事件处理模型）
private void initCommunicator(CommunicatorConfig config) throws CommunicatorConfigException {
//***** 省略代码 *****
this.threadPoolExecutor = ClientPoolManager.getClientThreadPoolExecutor(config);
//***** 省略代码 *****
}
public class ClientPoolManager {
public static ThreadPoolExecutor getClientThreadPoolExecutor(CommunicatorConfig communicatorConfig) {
//***** 省略代码 *****
clientThreadPoolMap.put(communicatorConfig, createThreadPool(communicatorConfig));
//***** 省略代码 *****
return clientPoolExecutor;
}
private static ThreadPoolExecutor createThreadPool(CommunicatorConfig communicatorConfig) {
int corePoolSize = communicatorConfig.getCorePoolSize();
int maxPoolSize = communicatorConfig.getMaxPoolSize();
int keepAliveTime = communicatorConfig.getKeepAliveTime();
int queueSize = communicatorConfig.getQueueSize();
TaskQueue taskqueue = new TaskQueue(queueSize);
String namePrefix = "tars-client-executor-";
TaskThreadPoolExecutor executor = new TaskThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.SECONDS, taskqueue, new TaskThreadFactory(namePrefix));
taskqueue.setParent(executor);
return executor;
}
}
（远程调用流程）
（底层代码写IO过程）
具体Reactor逻辑见上文2.5 网络模型内容，如果Reactor检查条件发现可以写IO的话也就是key.isWritable()为true，那么最终会循环从TCPSession.queue中取出ByteBuffer对象，调用SocketChannel.write(byteBuffer)执行实际的写网络Socket操作，代码逻辑见源码中的doWrite()方法。
读写事件线程池初始化
public class TCPSession extends Session {
public void write(Request request) throws IOException {
try {
IoBuffer buffer = selectorManager.getProtocolFactory().getEncoder().encodeRequest(request, this);
write(buffer);
//***** 省略代码 *****
}
protected void write(IoBuffer buffer) throws IOException {
//***** 省略代码 *****
if (!this.queue.offer(buffer.buf())) {
throw new IOException("The session queue is full. [ queue size:" + queue.size() + " ]");
}
if (key != null) {
key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
key.selector().wakeup();
}
}
protected synchronized int doWrite() throws IOException {
int writeBytes = 0;
while (true) {
ByteBuffer wBuf = queue.peek();
//***** 省略代码 *****
int bytesWritten = ((SocketChannel) channel).write(wBuf);
//***** 省略代码 *****
return writeBytes;
}
}
public class ServantClient {
public <T extends ServantResponse> T invokeWithSync(ServantRequest request) throws IOException {
//***** 省略代码 *****
ticket = TicketManager.createTicket(request, session, this.syncTimeout);
Session current = session;
current.write(request);
if (!ticket.await(this.syncTimeout, TimeUnit.MILLISECONDS)) {
//***** 省略代码 *****
response = ticket.response();
//***** 省略代码 *****
return response;
//***** 省略代码 *****
return response;
}
}
如代码所示，在执行完session.write()操作后，紧接着执行ticket.await()方法，该方法线程等待直到远程服务回复返回结果到客户端，ticket.await()被唤醒后，将执行后续操作，最终invokeWithSync方法返回response对象。其中Ticket的等待唤醒功能内部采用java.util.concurrent.CountDownLatch来实现。
对于异步方法调用，将会执行ServantClient.invokeWithAsync方法，也会创建一个Ticket，并且执行Session.write()操作，虽然不会调用ticket.await()，但是在Reactor接收到远程回复时，首先会先解析Tars协议头得到Response对象，然后将Response对象放入如图（Tars-Java的网络事件处理模型）所示的IO读写线程池中进行进一步处理，如下述源码（异步回调事件处理）所示，最终会调用WorkThread.run()方法，在run()方法里执行ticket.notifyResponse(resp)，该方法里面会执行类似上述代码2.1中的实现TarsAbstractCallback接口的调用成功回调的方法。
异步回调事件处理
public final class WorkThread implements Runnable {
public void run() {
try {
//***** 省略代码 *****
Ticket<Response> ticket = TicketManager.getTicket(resp.getTicketNumber());
//***** 省略代码 *****
ticket.notifyResponse(resp);
ticket.countDown();
TicketManager.removeTicket(ticket.getTicketNumber());
}
//***** 省略代码 *****
}
}
如下述源码所示，TicketManager会有一个定时任务轮训检查所有的调用是否超时，如果(currentTime - t.startTime) > t.timeout条件成立，那么会调用t.expired()告知回调对象，本次调用超时。
调用超时事件处理
public class TicketManager {
//***** 省略代码 *****
static {
executor.scheduleAtFixedRate(new Runnable() {
long currentTime = -1;
public void run() {
Collection<Ticket<?>> values = tickets.values();
currentTime = System.currentTimeMillis();
for (Ticket<?> t : values) {
if ((currentTime - t.startTime) > t.timeout) {
removeTicket(t.getTicketNumber());
t.expired();
}
}
}
}, 500, 500, TimeUnit.MILLISECONDS);
}
}
作者：vivo 互联网服务器团队-Ke Shengkai
我真的很习惯使用Ruby编写以下代码:my_hash={}my_hash['test']=1Java中对应的数据结构是什么？ 最佳答案 HashMapmap=newHashMap();map.put("test",1);我假设？ 关于java-等价于Java中的RubyHash，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.com/questions/22737685/
我正在尝试使用boilerpipe来自JRuby。我看过guide从JRuby调用Java，并成功地将它与另一个Java包一起使用，但无法弄清楚为什么同样的东西不能用于boilerpipe。我正在尝试基本上从JRuby中执行与此Java等效的操作:URLurl=newURL("http://www.example.com/some-location/index.html");Stringtext=ArticleExtractor.INSTANCE.getText(url);在JRuby中试过这个:require'java'url=java.net.URL.new("http://www
我只想对我一直在思考的这个问题有其他意见，例如我有classuser_controller和classuserclassUserattr_accessor:name,:usernameendclassUserController//dosomethingaboutanythingaboutusersend问题是我的User类中是否应该有逻辑user=User.newuser.do_something(user1)oritshouldbeuser_controller=UserController.newuser_controller.do_something(user1,user2)我
什么是ruby的rack或python的Java的wsgi？还有一个路由库。 最佳答案 来自Python标准PEP333:Bycontrast,althoughJavahasjustasmanywebapplicationframeworksavailable,Java's"servlet"APImakesitpossibleforapplicationswrittenwithanyJavawebapplicationframeworktoruninanywebserverthatsupportstheservletAPI.ht
一、引擎主循环UE版本：4.27一、引擎主循环的位置：Launch.cpp:GuardedMain函数二、、GuardedMain函数执行逻辑：1、EnginePreInit：加载大多数模块int32ErrorLevel=EnginePreInit(CmdLine);PreInit模块加载顺序：模块加载过程：（1）注册模块中定义的UObject，同时为每个类构造一个类默认对象（CDO，记录类的默认状态，作为模板用于子类实例创建）（2）调用模块的StartUpModule方法2、FEngineLoop::Init()1、检查Engine的配置文件找出使用了哪一个GameEngine类（UGame
这篇文章是继上一篇文章“Observability：从零开始创建Java微服务并监控它（一）”的续篇。在上一篇文章中，我们讲述了如何创建一个Javaweb应用，并使用Filebeat来收集应用所生成的日志。在今天的文章中，我来详述如何收集应用的指标，使用APM来监控应用并监督web服务的在线情况。源码可以在地址 https://github.com/liu-xiao-guo/java_observability 进行下载。摄入指标指标被视为可以随时更改的时间点值。当前请求的数量可以改变任何毫秒。你可能有1000个请求的峰值，然后一切都回到一个请求。这也意味着这些指标可能不准确，你还想提取最小/
HashMap中为什么引入红黑树，而不是AVL树呢1.概述开始学习这个知识点之前我们需要知道，在JDK1.8以及之前，针对HashMap有什么不同。JDK1.7的时候，HashMap的底层实现是数组+链表JDK1.8的时候，HashMap的底层实现是数组+链表+红黑树我们要思考一个问题，为什么要从链表转为红黑树呢。首先先让我们了解下链表有什么不好？？？2.链表上述的截图其实就是链表的结构，我们来看下链表的增删改查的时间复杂度增：因为链表不是线性结构，所以每次添加的时候，只需要移动一个节点，所以可以理解为复杂度是N(1)删：算法时间复杂度跟增保持一致查：既然是非线性结构，所以查询某一个节点的时候
遍历文件夹我们通常是使用递归进行操作，这种方式比较简单，也比较容易理解。本文为大家介绍另一种不使用递归的方式，由于没有使用递归，只用到了循环和集合，所以效率更高一些！一、使用递归遍历文件夹整体思路1、使用File封装初始目录，2、打印这个目录3、获取这个目录下所有的子文件和子目录的数组。4、遍历这个数组，取出每个File对象4-1、如果File是否是一个文件，打印4-2、否则就是一个目录，递归调用代码实现publicclassSearchFile{publicstaticvoidmain(String[]args){//初始目录Filedir=newFile("d:/Dev");Datebeg
我基本上来自Java背景并且努力理解Ruby中的模运算。(5%3)(-5%3)(5%-3)(-5%-3)Java中的上述操作产生，2个-22个-2但在Ruby中，相同的表达式会产生21个-1-2.Ruby在逻辑上有多擅长这个？模块操作在Ruby中是如何实现的？如果将同一个操作定义为一个web服务，两个服务如何匹配逻辑。 最佳答案 在Java中，模运算的结果与被除数的符号相同。在Ruby中，它与除数的符号相同。remainder()在Ruby中与被除数的符号相同。您可能还想引用modulooperation.
Java的Collections.unmodifiableList和Collections.unmodifiableMap在Ruby标准API中是否有等价物？ 最佳答案 使用freeze应用程序接口(interface):Preventsfurthermodificationstoobj.ARuntimeErrorwillberaisedifmodificationisattempted.Thereisnowaytounfreezeafrozenobject.SeealsoObject#frozen?.Thismethodretur