Spring 声明式事务中进行多线程操作及解析

今天项目还存在一个问题，而明天就要上线了，所以很急，主要的问题如下：

 

问题描述：主要的业务逻辑--页面上处理放行基金下单记录（主管审核以后的操作），这时候需要做的动作如下：

 

1. 把下单记录关联的档案移动到MFT（TIBCO公司产品，文档传送软件）的发送目录；

2. 更新下单记录信息（OrderInfo）的状态；

3. Daemon（JAVA注册的windows service）程序去检测MFT发送目录是否有档案，有则起一个新的Thread调用MFT Server去发送档案，MFT会自动发送到目标目录，发送成功或失败会有一个回覆信息，AP端根据这个状态去更新下单记录（OrderInfo）的状态；

 

处理以上业务逻辑的service类使用了Spring声明式事务进行管理，然后客户在测试机上反馈的信息是：

1. 抛出一下异常信息

 INFO [WebContainer : 4] [2010-10-25 17:17:01,359] (com.service.impl.OrderCenterImpl) - Release Order! [uniqueNo:94101025C,orderNo:1]
 WARN [WebContainer : 4] [2010-10-25 17:17:03,531] (org.hibernate.util.JDBCExceptionReporter) - SQL Error: 1222, SQLState: S00051
ERROR [WebContainer : 4] [2010-10-25 17:17:03,531] (org.hibernate.util.JDBCExceptionReporter) - 鎖定要求的逾時期間已過。
ERROR [WebContainer : 4] [2010-10-25 17:17:03,546] (com.service.impl.FunctionActionImpl) - Hibernate operation: could not execute update query; bad SQL grammar [update SF_OrderMessages set status=? where TransactionNo=? and msgLevel=?]; nested exception is com.microsoft.sqlserver.jdbc.SQLServerException: 鎖定要求的逾時期間已過。
org.springframework.jdbc.BadSqlGrammarException: Hibernate operation: could not execute update query; bad SQL grammar [update OrderInfo set status=? where id=?]; nested exception is com.microsoft.sqlserver.jdbc.SQLServerException: 鎖定要求的逾時期間已過。
com.microsoft.sqlserver.jdbc.SQLServerException: 鎖定要求的逾時期間已過。
	at com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDatabaseError(Unknown Source)
	at com.microsoft.sqlserver.jdbc.SQLServerStatement.getNextResult(Unknown Source)
	at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.doExecutePreparedStatement(Unknown Source)
	at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement$PrepStmtExecCmd.doExecute(Unknown Source)

 

 WARN [WebContainer : 4] [2010-10-25 17:17:03,859] (org.hibernate.util.JDBCExceptionReporter) - SQL Error: 1205, SQLState: 40001
org.springframework.dao.DeadlockLoserDataAccessException: Hibernate operation: could not execute update query; SQL [update OrderInfo set status=? where id=?]; 交易 (處理序識別碼 134) 在 鎖定 資源上被另一個處理序鎖死並已被選擇作為死結的犧牲者。請重新執行該交易。; nested exception is com.microsoft.sqlserver.jdbc.SQLServerException: 交易 (處理序識別碼 134) 在 鎖定 資源上被另一個處理序鎖死並已被選擇作為死結的犧牲者。請重新執行該交易。
com.microsoft.sqlserver.jdbc.SQLServerException: 交易 (處理序識別碼 134) 在 鎖定 資源上被另一個處理序鎖死並已被選擇作為死結的犧牲者。請重新執行該交易。
	at com.microsoft.sqlserver.jdbc.SQLServerException.makeFromDatabaseError(Unknown Source)
	at com.microsoft.sqlserver.jdbc.SQLServerStatement.getNextResult(Unknown Source)
	at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement.doExecutePreparedStatement(Unknown Source)
	at com.microsoft.sqlserver.jdbc.SQLServerPreparedStatement$PrepStmtExecCmd.doExecute(Unknown Source)
	at com.microsoft.sqlserver.jdbc.TDSCommand.execute(Unknown Source)

 

2. MFT的档案没有传送到指定的目录，即传送失败

 

分析原因：

1. 在声明式事务里使用多线程去更新OrderInfo表（主要原因）

2. Daemon也去更新OrderInfo表

 

MFTSenderServiceImpl.java在Spring里配置了声明式事务管理，其transfer方法中迭代调用transferFile方法：

	public void transfer() throws Exception {
		
		.....
		.....
		
		for (int i = 0; i < movedFiles.size(); i++) {
			transferFile(movedFiles.get(i));
		}
	}

  

	private void transferFile(final String fileName) throws Exception {
		...
		...
		mftSenter.sendFiles(faxFilePaths.toArray(new String[] {}), new SenderCallback() {

			public void failed() {
				...
				//update OrderInfo operation
			}

			public void success() {
				...
				//update OrderInfo operation
　　　　　　　　　　　　　　　　 }
		});
		...
		...
	}

 

	public void sendFiles(String[] files, SenderCallback callBack) {
		sendMFTFile(faxAgentName, faxAgentDir, files, mftCLIcmd, maxRetry, callBack);
	}

	public static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 3, TimeUnit.SECONDS,
			new ArrayBlockingQueue<Runnable>(50), new RejectedExecutionHandler() {

				public void rejectedExecution(final Runnable r, final ThreadPoolExecutor executor) {
					Thread thread = new Thread() {
						public void run() {
							try {
								Thread.sleep(1000);
							} catch (InterruptedException e) {
							}
							logger.debug(r.toString() + ":　readd current thread to threadPool");
							executor.execute(r);
						};
					};
					thread.start();
				}
			});

	protected void sendMFTFile(String tdccAgentName, String tdccAgentDir, String[] pathes, String mftCLIcmd,
			int maxRetry, final SenderCallback callback) {
			...
			...
			MFTSendThread sendThread = new MFTSendThread(tdccAgentName, tdccAgentDir, newPathes
					.toArray(new String[] {}), mftCLIcmd, maxRetry, callback);
			threadPool.execute(sendThread);
	}

 

用到了JDK的并发处理ThreadPoolExecutor，并提供了自定义的RejectedExecutionHandler的处理，关于ThreadPoolExecutor的几个参数的解释如下：

1. corePoolSize: 线程池中保持的核心线程数（可以认为是最少线程数）

2. maximumPoolSize: 线程池最大线程数

3. keepAliveTime: 某线程idle时间超过该时间，则自动terminate

4. unit: 时间单位

5. workQueue: 线程池的工作队列

6. handler: 如果线程池中的线程数超过了workQueue的容量（当然肯定也超过core的容量）进行的后续处理

 

解析： 当线程池建立以后，如果执行execute(Runable r);方法添加一个线程时，根据以下情况进行处理：

(1) 如果curSize<corePoolSize，即使池中存在idle状态的线程，也创建一个线程补充到池中；

(2) 如果curSize=corePoolSize，且workQueue未满，则加入到workQueue中；

(3) 如果curSize>corePoolSize，而workQueue已满且curSize<maximumPoolSize，创建一个线程加入到池中；

(4) 如果curSize>corePoolSize，而workQueue已满且curSize>=maximumPoolSize，使用handler处理，默认JDK对handler提供了几种实现，根据实际要求选择或者自行实现RejectedExecutionHandler类的rejectedExecution方法。

 

curSize: 当前线程池的数量

 

那么：就出现了事务中使用多线程去更新DB中某个table中的多条记录的情况，造成死锁的情况就很容易出现了。

 

<?xml version="1.0"?>
<!DOCTYPE beans PUBLIC "-//SPRING//DTD BEAN//EN" "http://www.springframework.org/dtd/spring-beans.dtd">

<beans>
	<bean id="faxCenterImpl" class="com..service.impl.FaxCenterImpl">
		....
	</bean>

	<bean id="faxCenter" parent="transProxy">
		<property name="target" ref="faxCenterImpl" />
	</bean>
</beans>

 

最后的解决方式是，Daemon程序没有采用事务处理的bean实例faxCenter，而是使用faxCenterImpl的实例，该实例没有纳入Spring声明式事务管理，而其他地方用到的faxCenter依然是有事务管理的实例对象。

 

public class FaxTransferProcess extends BaseDaemonInSpringContext {

	private static final Log logger = LogFactory.getLog(FaxTransferProcess.class);

	private IFaxCenter faxCenter;

	public void init(ApplicationContext context) {
		faxCenter = (IFaxCenter) context.getBean("faxCenterImpl");//faxCenterImpl而不是faxCenter
	}

	public void process() {
		try {
			faxCenter.transfer();
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}
}

 

这样保证了在事务里不会去多线程更新记录是比较好的解决方式，看网上一些讨论的做法是利用hibernate的锁机制来处理多线程的问题，我没有去试过，也不知道好不好。但是至少这个问题是解决了，以后有空再去研究一下事务中并发处理的问题。

 

【参考文章】

1. http://mysaga.iteye.com/blog/436718

2. http://www.iteye.com/topic/515634

评论

4 楼 can4you 2011-05-20  

youjianbo_han_87 写道

对同一个表做update操作时，多线程能起作用吗？ update操作应该会有行级锁把。。。

我这里的解决方式是把更新数据的processor类不纳入事务管理。

引用

  
<bean id="faxCenterImpl" class="com..service.impl.FaxCenterImpl"> 
    ....  
</bean> 
<bean id="faxCenter" parent="transProxy"> 
   <property name="target" ref="faxCenterImpl" /> 
</bean> 

这里faxCenter是纳入了事务管理的，但是我在processor中取的bean的时候使用：

引用

faxCenter = (IFaxCenter) context.getBean("faxCenterImpl");//faxCenterImpl而不是faxCenter

我的理解是，存在事务，然后多线程更新，在事务提交时，很容易造成死锁（就是因为存在行级锁）。
现在不纳入事务管理，每个线程去更新数据后立刻提交，造成死锁的概率下降了。这里只是折中的解决方式。

3 楼 youjianbo_han_87 2011-05-04  

对同一个表做update操作时，多线程能起作用吗？ update操作应该会有行级锁把。。。

2 楼 can4you 2010-12-22  

哈哈，是啊，这都被你发现啦。。。。

1 楼 gelnyang 2010-12-21  

Iain？