A lot depends on the timeout limit of your database transactions… All activiti-calls are done within a database-transation. If the operation takes too long, the transaction will be timed out and exception is thrown (not marking the job as executed and unlocking it to be acquired again by the job-executor).
On top of that, to prevent job-executor not to block all of it's threads too long, a max-lock time is set (default is 5 minutes: int lockTimeInMillis = 5 * 60 * 1000
. If expired, job executor presumes something went wrong and unlocks the job, effectively marking it to be executed again.
The lock-time is a setter on the JobExecutor and can be altered in configuration or even from code, so this can be raised.
But consider that all job-executions are executed in a transaction. Having a lot of long-running transactions can influence db-performance. If you don't actually need to update process-properties very often when the async-servicetask is running, I suggest working with a thread-pool outside of activiti:
On top of that, to prevent job-executor not to block all of it's threads too long, a max-lock time is set (default is 5 minutes: int lockTimeInMillis = 5 * 60 * 1000
. If expired, job executor presumes something went wrong and unlocks the job, effectively marking it to be executed again. The lock-time is a setter on the JobExecutor and can be altered in configuration or even from code, so this can be raised.
But consider that all job-executions are executed in a transaction. Having a lot of long-running transactions can influence db-performance. If you don't actually need to update process-properties very often when the async-servicetask is running, I suggest working with a thread-pool outside of activiti:
- The service-task prepares (eg. read variables from process) an creates the "job" or runnable to be executed by the external thread pool
- Service-task queues the job (saving reference to process-instance id) and finishes
- Right after the service-taks, there's a receive-task that waits, releasing the calling thead and not holding open the transaction.
- Long-running operation completes and signals the process-instance and perhaps set some variables on the process based on the results before signaling.
Thanks again for the valuable information. You have answered all my questions fully. This forum has been a deployment-saver 
.
I was hoping you could do a quick review of my Service Task and Runnable to verify that i'm doing things correctly. As of now, i've taken your suggestion on using the Java Receive Task and an asynchronous thread. I've pasted my code below (Of course, i've stolen activiti's logic for running a shell task 😞
thanks so much for your continued help. I am learning lots!
.I was hoping you could do a quick review of my Service Task and Runnable to verify that i'm doing things correctly. As of now, i've taken your suggestion on using the Java Receive Task and an asynchronous thread. I've pasted my code below (Of course, i've stolen activiti's logic for running a shell task
😞
public class ShellServiceTask implements JavaDelegate {
final static Logger logger = Logger.getLogger(ShellServiceTask.class.getName());
protected Expression command;
protected Expression wait;
protected Expression arg1;
protected Expression arg2;
protected Expression arg3;
protected Expression arg4;
protected Expression arg5;
protected Expression outputVariable;
protected Expression errorCodeVariable;
protected Expression redirectError;
protected Expression cleanEnv;
protected Expression directory;
String commandStr;
String arg1Str;
String arg2Str;
String arg3Str;
String arg4Str;
String arg5Str;
String waitStr;
String resultVariableStr;
String errorCodeVariableStr;
Boolean waitFlag;
Boolean redirectErrorFlag;
Boolean cleanEnvBoolan;
String directoryStr;
@Override
public void execute(DelegateExecution execution) throws Exception {
logger.info("In " + command.getValue(execution) + " ShellTask script with variables = " + execution.getVariables());
executeScript(execution);
// create a new value for execution
logger.info("Leaving " + command.getValue(execution) + " ShellTask script with variables = " + execution.getVariables());
}
private void readFields(DelegateExecution execution) {
commandStr = getStringFromField(command, execution);
arg1Str = getStringFromField(arg1, execution);
arg2Str = getStringFromField(arg2, execution);
arg3Str = getStringFromField(arg3, execution);
arg4Str = getStringFromField(arg4, execution);
arg5Str = getStringFromField(arg5, execution);
waitStr = getStringFromField(wait, execution);
resultVariableStr = getStringFromField(outputVariable, execution);
errorCodeVariableStr = getStringFromField(errorCodeVariable, execution);
String redirectErrorStr = getStringFromField(redirectError, execution);
String cleanEnvStr = getStringFromField(cleanEnv, execution);
waitFlag = waitStr == null || waitStr.equals("true");
redirectErrorFlag = redirectErrorStr != null && redirectErrorStr.equals("true");
cleanEnvBoolan = cleanEnvStr != null && cleanEnvStr.equals("true");
directoryStr = getStringFromField(directory, execution);
}
public String toString() {
return "command="+command+", arg1="+arg1+", arg2="+arg2+", arg3="+arg3+", arg4="+arg4+", arg5="+arg5+
", wait="+wait+", outputVariable="+outputVariable+", errorCodeVariable="+errorCodeVariable;
}
public void executeScript(DelegateExecution execution) {
logger.info("starting execute");
readFields(execution);
List<String> argList = new ArrayList<String>();
argList.add(commandStr);
if (arg1Str != null)
argList.add(arg1Str);
if (arg2Str != null)
argList.add(arg2Str);
if (arg3Str != null)
argList.add(arg3Str);
if (arg4Str != null)
argList.add(arg4Str);
if (arg5Str != null)
argList.add(arg5Str);
logger.info("Firing executor with processInstanceId = " + execution.getProcessInstanceId());
Thread executor = new Thread(new ShellTaskThread(argList, waitFlag, redirectErrorFlag, cleanEnvBoolan, directoryStr,
resultVariableStr, errorCodeVariableStr, execution.getProcessInstanceId()));
executor.start();
logger.info("Executor started…leaving workflow");
}
protected String getStringFromField(Expression expression, DelegateExecution execution) {
if (expression != null) {
Object value = expression.getValue(execution);
if (value != null) {
return value.toString();
}
}
return null;
}
}
class ShellTaskThread implements Runnable {
final static Logger logger = Logger.getLogger(ShellTaskThread.class.getName());
List<String> commands;
boolean waitFlag;
String resultVariableStr;
String errorCodeVariableStr;
boolean cleanEnv;
boolean redirectErrorFlag;
String directoryStr;
String processId;
public ShellTaskThread(List<String> commands, boolean waitFlag, boolean redirectErrorFlag,
boolean cleanEnv, String directoryStr,
String resultVariableStr, String errorCodeVariableStr, String processId) {
this.commands = commands;
this.waitFlag = waitFlag;
this.resultVariableStr = resultVariableStr;
this.errorCodeVariableStr = errorCodeVariableStr;
this.redirectErrorFlag = redirectErrorFlag;
this.cleanEnv = cleanEnv;
this.directoryStr = directoryStr;
this.processId = processId;
}
@Override
public void run() {
logger.info("running async thread with command: " + commands);
logger.info("other vars: resultVariableStr: " + resultVariableStr + " errorCodeVariableStr: " +
errorCodeVariableStr + " processId: " + processId);
// set activiti context
ProcessEngine processEngine = ProcessEngines.getDefaultProcessEngine();
RuntimeService runtimeService = processEngine.getRuntimeService();
Execution execution = runtimeService.createExecutionQuery().processInstanceId(processId).singleResult();
logger.info("execution = " + execution.getId() + ": " + execution.getProcessInstanceId());
ProcessBuilder processBuilder = new ProcessBuilder(commands);
try {
processBuilder.redirectErrorStream(redirectErrorFlag);
if (cleanEnv) {
Map<String, String> env = processBuilder.environment();
env.clear();
}
if (directoryStr != null && directoryStr.length() > 0)
processBuilder.directory(new File(directoryStr));
Process process = processBuilder.start();
if (waitFlag) {
try {
int errorCode = process.waitFor();
logger.info("errorCode = " + errorCode);
if (resultVariableStr != null) {
String result = convertStreamToStr(process.getInputStream());
runtimeService.setVariable(execution.getId(), resultVariableStr, result);
}
if (errorCodeVariableStr != null) {
runtimeService.setVariable(execution.getId(), errorCodeVariableStr, Integer.toString(errorCode));
}
} catch (InterruptedException e) {
if (resultVariableStr != null) {
runtimeService.setVariable(execution.getId(), resultVariableStr, e);
}
if (errorCodeVariableStr != null) {
runtimeService.setVariable(execution.getId(), errorCodeVariableStr, "1");
}
throw e;
}
}
} catch (Exception e) {
logger.info("Could not execute shell command " + commands + " with error = " + e);
} finally {
logger.info("Leaving thread with process variables = " + runtimeService.getVariables(execution.getId()));
runtimeService.signal(execution.getId());
}
}
public static String convertStreamToStr(InputStream is) throws IOException {
if (is != null) {
Writer writer = new StringWriter();
char[] buffer = new char[1024];
try {
Reader reader = new BufferedReader(new InputStreamReader(is, "UTF-8"));
int n;
while ((n = reader.read(buffer)) != -1) {
writer.write(buffer, 0, n);
}
} finally {
is.close();
}
return writer.toString();
} else {
return "";
}
}
}thanks so much for your continued help. I am learning lots!
