DActor

Introduction

DActor框架基于協程思想設計，可同時支持同步和異步代碼，簡化在線異步代碼的開發，用同步代碼的思維來開發異步代碼，兼顧異步代碼的高并發、無阻塞和同步代碼的易讀性，可維護性。

最大程度的降低阻塞，提高單個線程的處理能力，并可有效的降低線程數。

項目地址

文檔

視圖配置說明:視圖配置說明

SpringBoot Yml 配置說明:Yml配置說明

XML 配置說明:XML配置說明

常用類說明:常用類說明

配置和API說明:配置和API說明

更多文檔詳見：WIKI

QQ交流群

783580303

接入系統

Dpress：基于Dactor構建的多域名博客系統

持續更新中。。歡迎大家自薦

Overview

目前開發過程中的幾個常見模型

同步編程

所有步驟都在一個主線程中完成,調用一個方法，等待其響應返回。一個請求占用一個線程,在有數據庫操作、TCP和Http通訊時因為有阻塞情況，會導致占用線程占用而無法及時釋放

，因此在同步交易中引入了線程池概念,提高系統的吞吐量

異步編程

所有步驟都可在不同線程中完成,調用一個方法，不等待響應既返回，典型交易如NodeJs。

目前市面上的異步框架都比較復雜，市面的通用解決方案是CallBack和Promise/Deferred模式模式。

設計思路

為了保留異步的高性能,簡化異步的開發模式，同時使得程序更容易被程序員理解，在性能和代碼可閱讀性中間取得平衡，設計了此框架。

處理步驟:將請求封裝為消息,丟入消息隊列，尋找合適步驟處理消息,上述過程不斷循環，直到所有可用步驟都執行完畢。

因為是對消息隊列進行處理,對于同步交易，處理完畢即可丟入消息隊列。對于異步交易，等待回調完畢再丟入消息隊列。

兩種情況對于框架來說是無差別的。同時因為通過異步交易避免了阻塞情況的發生,所以可在不大幅度提高線程數的情況下，提高吞吐量，

同時也可在一定程度避免流量突增的情況發生。

消息隊列采用Disruptor的的高性能隊列RingBuffer。

以Actor協程并發模型為基礎設計框架。

Features

1、集成Netty

2、集成HttpClient

3、集成HttpServlet

4、支持多層父子結構

5、支持責任鏈模式

6、J2EE支持json,csv,pdf,xml,html格式輸出

7、J2EE支持數據流輸出,動態文件下載、動態圖片輸出、跳轉和可根據配置動態輸出

8、同時支持SpringBoot和Spring

環境要求

JDK 1.8

Spring FrameWork 5.2.4.RELEASE + 或者 Spring Boot 2.2.7.RELEASE +

Servlet 3.0+(因為需要使用Servlet的異步功能)

注意事項

請求的完整邏輯是分散在不同的線程中執行的，所以盡量避免使用ThreadLocal

Getting Started

example是J2EE程序,下載后，可直接運行,其中集成了若干例子

默認使用.do提交相關交易，但如果是.json將會返回json數據

啟動后,在瀏覽器中輸入http://localhost:8080/example/randomTxt2.json

輸出的是json格式的字符串

randomTxt2:只有一級父子關系

randomTxt1:有二級父子關系

chaintest1:只使用責任鏈

chaintest2:同時使用責任鏈和一級父子關系

exceptionTest:子交易拋出錯誤,框架對錯誤的處理

randomTxt3為beginBeanId為Actor標簽的BeanId例子

httptest演示的是通過httpclient異步方式訪問百度網站

訪問URL:http://localhost:8080/example/ httptest.do

http://localhost:8080/example/np.randomTxt2.json為使用命名空間的例子,相關配置在conf/namespace.xml中。

啟動后，可在控制臺看到內部調用結果

Maven dependency

cn.ymotel

dactor

1.1.2

Gradle dependency

compile group: 'cn.ymotel', name: 'dactor', version:'1.1.2'

代碼簡單講解

執行過程為chain->grandfather->parent->Selft。

依次調用執行責任鏈中邏輯，grandfather中的邏輯，parent的邏輯和自身邏輯。

chain,grandfather,parent都可為空，不設置

在grandfather和parent中的Steps中至少有一個為placeholderActor交易，以調用子邏輯

整個過程中，需要先設置全局占位符

交易中如果未填寫beginBeanId或者endBeanId時，系統默認使用全局中配置的beginBeanId或者endBeanId

condtion可為空，空字符串，或者是ognl表達式

placeholderActor的作用是在暫存當前環境，并調用子交易，待子交易執行完畢后，再恢復當前環境繼續執行

如果在Step中未找到toBeanIdActor，會直接調用endBeanId方法，認為自身交易已執行結束。

交易的請求和流轉信息都保存在Message中

如果指定handleException=false或者使用默認設置，直接返回父中執行，如果父中也未捕獲，則繼續返回上一級執行，

一般來說至少有要有一個actor中指定handleException=true

啟動框架接收和執行請求

同步異步寫法

同步寫法

public class BeginActor implements Actor {

/* (non-Javadoc)

* @see com.ymotel.util.actor.Actor#HandleMessage(com.ymotel.util.actor.Message)

*/

@Override

public Object HandleMessage(Message message) throws Exception {

return message;

}

}

繼承Actor接口，在HandlerMessage返回message方法，框架判斷返回不為空的情況下，直接將結果丟給框架進行下一步出來

異步寫法

以HttpClientActor為例,在HandlerMessage中最后返回null，框架收到請求后等待下一步處理，釋放線程，在HttpClient中的CallBack中調用

message.getControlMessage().getMessageDispatcher().sendMessage(message);框架收到后進行下一步處理。

public Object HandleMessage(final Message message) throws Exception {

Map context = message.getContext();

//

HttpUriRequest request = getHttpBuild(message.getContext()).build();

if (referer != null) {

request.addHeader("Referer", referer);

}

/**

* 通過此處可共用會話，進行類似登錄后交易

*/

HttpClientContext tmplocalContext = null;

if (context.containsKey(HTTPCLIENT_CONTEXT)) {

tmplocalContext = (HttpClientContext) context.get(HTTPCLIENT_CONTEXT);

} else {

tmplocalContext = HttpClientContext.create();

CookieStore cookieStore = new BasicCookieStore();

tmplocalContext.setCookieStore(cookieStore);

context.put(HTTPCLIENT_CONTEXT, tmplocalContext);

}

final HttpClientContext localContext = tmplocalContext;

// CookieStore cookieStore = new BasicCookieStore();

// localContext.setCookieStore(cookieStore);

if (logger.isInfoEnabled()) {

logger.info("HandleMessage(Message) - httpclient----" + request); //$NON-NLS-1$

}

// final HttpGet httpget = new HttpGet(uri);

final String tmpcontent = content;

final String tmpcharset = charset != null ? charset : (String) context.get(CHARSET);

httpClientHelper.getHttpclient().execute(request, localContext, new FutureCallback() {

public void completed(final HttpResponse response) {

try {

/**

* 完成后及時清除

*/

message.getContext().remove(tmpcontent);

actorHttpClientResponse.handleResponse(response, localContext, tmpcharset, message);

//String responseString=HandleResponse((String)message.getContext().get(CHARSET),response);

//message.getContext().put(RESPONSE, responseString);

} catch (Exception e) {

// TODO Auto-generated catch block

if (logger.isErrorEnabled()) {

logger.error("$FutureCallback.completed(HttpResponse)", e); //$NON-NLS-1$

}

message.setException(e);

message.getControlMessage().getMessageDispatcher().sendMessage(message);

}

// System.out.println(httpget.getRequestLine() + "->" + response.getStatusLine());

}

public void failed(final Exception ex) {

if (logger.isErrorEnabled()) {

logger.error("$FutureCallback.failed(Exception)", ex); //$NON-NLS-1$

}

message.setException(ex);

message.getControlMessage().getMessageDispatcher().sendMessage(message);

// System.out.println(httpget.getRequestLine() + "->" + ex);

}

public void cancelled() {

Exception exception = new Exception("已取消");

message.setException(exception);

message.getControlMessage().getMessageDispatcher().sendMessage(message);

// System.out.println(httpget.getRequestLine() + " cancelled");

}

});

return null;

}

配置和API說明

配置說明

通過在xml中的Step實現內部Actor之間的流程跳轉

在配置文件中包含

Actor、chain、和global配置 。

程序整個執行順序為根據交易碼找到對應的Actor,然后執行按照chain->parent->selft的順序進行執行。

chain執行到placeholder處,調用parent交易繼續執行,在parent交易中執行到placeholder交易后，調用selft自身交易繼續執行。

自身交易執行完畢，彈出parent的placeholder處交易繼續執行.parent執行完畢,彈出chain中代碼繼續執行。

global配置如下

beginBeanId為默認的開始Actor,value中的值是在Spring中對應的beanName，程序初始化時將會取得此值，對未指定beginBeanId或者endBeanId的Actor初始化全局配置。

beginActor和endActor都需要繼承Actor接口。

actor配置如下

htmlstream:

屬性handleException如果不設置的話,遇到異常,程序將會認為子類中已經執行完畢,跳到parent中PlaceHolder處執行。設置為true，將不會直接跳轉到parent中，由子類進行自我處理。

parent和chain為調用具體交易前需要調用的公共交易，由于大部分交易都有通用的前置交易和統一的后置交易。通過設置parent或者chain，可提高代碼復用度。

fromBeanId和toBeanId配置的是Actor或者實現Actor接口的beanId。

parent和chain中的ref都需要是Actor.

results中可定義返回的state和需要處理的viewActor

async標記是否是旁路交易，默認值為false，為true值時,會將上下文內容設置復制一份，重新生成一份Message,進行執行，不影響主流程。

chain配置

chain可直觀展現Actor調用順序.

在chain中可順序并列多個parent類。每個parent中的Step都需要有placeHolderActor，以調用子類。

依次執行list中的交易，再執行自身交易。自身交易執行完畢，再依次回溯責任鏈中的每個交易，直到無可用交易。

命名空間

xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

xmlns:aop="http://www.springframework.org/schema/aop"

xmlns:actor="http://www.ymotel.cn/schema/dactor"

xmlns:beans="http://www.springframework.org/schema/beans"

xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd

http://www.ymotel.cn/schema/dactor http://www.ymotel.cn/schema/dactor.xsd" namespace="np">

重要類方法說明

cn.ymotel.dactor.core.MessageDispatcher是交易流轉的核心接口類

public void startMessage(Message message, ActorTransactionCfg actorcfg, boolean blocked) throws Exception

方法,用于開始整個流程，其中message需要在執行前進行構造，actorcfg可通過spring的getBean方法得到為Actor對象，如下

通過getBean('randomTxt1')即可得到ActorTransactionCfg對象。

blocked為是否阻塞，一般在交易初次放入隊列是為false,表示如果隊列滿，則直接扔給客戶端進行處理。為true則一般為內部交易,必須提交給隊列進行處理。

sendMessage方法內部調用，用于將處理完畢的Message重新放入隊列，繼續下一步流程。

cn.ymotel.dactor.core.disruptor.MessageRingBufferDispatcher是MessageDispatcher的接口實現類。,在啟動Spring是需要在配置中加上

MessageRingBufferDispatcher的strategy、bufferSize、threadNumber為三個可設置屬性.正常情況下使用默認設置即可。

strategy默認使用ringBuffer的BlockingWaitStrategy策略進行調度，如果交易量比較大，可調整此策略。

bufferSize默認使用1024。

threadNumber默認使用CPU個數的線程數。

其他默認Actor說明

cn.ymotel.dactor.message.Message.Actor,所有需要在執行的交易都必須繼承此接口。

public Object HandleMessage(Message message) throws Exception;程序通過調用HandleMessage對象,如果返回的不是message對象或者為NULL,則認為此交易是異步執行,不再自行調度。由異步交易在收到請求后，自己調用將Message再此放入隊列中。

cn.ymotel.dactor.action.PlaceholderActor 交易為特殊交易，用來將當前隊列暫存，并調用子交易。

cn.ymotel.dactor.action.BeginActor 為Actor中step的默認開始交易。

cn.ymotel.dactor.action.EndActor 為Actor中step的默認結束交易。

cn.ymotel.dactor.action.JsonViewResolverActor為需要返回Json的J2EE view

cn.ymotel.dactor.action.ViewResolveActor為需要返回J2EE view的統一處理Actor

cn.ymotel.dactor.action.httpclient.HttpClientActor 提供的異步調用httpClient的Actor

cn.ymotel.dactor.action.netty.aysnsocket.TcpClientActor 提供的異步調用netty的Actor

交易流程舉例說明

以上交易的交易流程圖如下

以上的完整例子都可在example中得到

總結

以上是生活随笔為你收集整理的java 协程框架_GitHub - yaozhang0105/dactor: Dactor是基于Java的轻量级同步异步统一处理框架,基于协程思想构建...的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。