<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration"    xmlns:int-jpa="http://www.springframework.org/schema/integration/jpa"    xmlns:int-jms="http://www.springframework.org/schema/integration/jms"    xmlns:context="http://www.springframework.org/schema/context"    xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd            http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.2.xsd            http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd            http://www.springframework.org/schema/context    http://www.springframework.org/schema/context/spring-context.xsd                        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd            http://www.springframework.org/schema/integration/jpa http://www.springframework.org/schema/integration/jpa/spring-integration-jpa-2.2.xsd            ">        <int-jpa:inbound-channel-adapter        auto-startup="true" entity-manager="em"        send-timeout="60000" channel="process.channel"        expect-single-result="true"        jpa-query="SELECT sysdate FROM dual">        <int:poller fixed-delay="60000" />    </int-jpa:inbound-channel-adapter>            <int:channel id="process.channel">        <int:queue capacity="1"/>            </int:channel>    <int:chain input-channel="process.channel">                    <int-jpa:retrieving-outbound-gateway entity-manager="em" jpa-query="SELECT sp FROM SmsMessage sp Where sp.tatus is null order by sp.requestOn,sp.id"/>                            <int:splitter ref="process.processSplitter" method="split"/>                <int:service-activator ref="process.smsSenderService"            method="send" />                    <int:poller fixed-delay="5000" receive-timeout="-1"/>                </int:chain>                <bean id="process.smsSenderService" class="com.yd.core.service.SmsSenderService" />                <bean id="process.processSplitter" class="com.yd.core.service.PaymentProcessSplitter" /></beans>

import org.springframework.context.ApplicationContext;import org.springframework.integration.MessageChannel;import org.springframework.integration.support.MessageBuilder;public class JobWorker implements Runnable {    private static final int DEFAULT_WAIT_TIME = 3000;    @Override    public void run() {        while (true) {            try {                LoggerUtil.getJobLogger().info("JobWorker, Ready for take job run request.");                JobRunnerRequest jobRequest = JobManagerService.getJobManager().takeRequest();                while (jobRequest == null) {                    LoggerUtil.getJobLogger().warn("JobWorker, jobRequest is null, will try to get the job requet again.");                    Thread.sleep(DEFAULT_WAIT_TIME);                    jobRequest = JobManagerService.getJobManager().takeRequest();                }                LoggerUtil.getJobLogger().info("JobWorker, Received a job run request.");                MessageChannel channel = findChannel(jobRequest.getJobChannelId());                if (channel != null) {                    channel.send(MessageBuilder.withPayload(jobRequest.getJobMessagePayload()).build());                    LoggerUtil.getJobLogger().info("JobWorker, Completed to sned message to job channel");                }            }            catch (Exception ex) {                LoggerUtil.getJobLogger().warn("JobWorker, Completed to sned message to job channel");            }        }    }    private MessageChannel findChannel(String jobChannelId) {        ApplicationContext context = ApplicationContextProvider.getContext();        if (context == null) {            LoggerUtil.getJobLogger().error(String.format("JobWorker, Cannot get the application context, to startup job %s", jobChannelId));            return null;        }        Object channel = context.getBean(jobChannelId);        if (channel instanceof MessageChannel) {            return (MessageChannel) channel;        }        else {            LoggerUtil.getJobLogger().error(String.format("JobWorker, Cannot get the message bean, to startup job %s", jobChannelId));            return null;        }    }}

import java.util.concurrent.BlockingQueue;import java.util.concurrent.LinkedBlockingQueue;public final class JobManagerService {    private BlockingQueue<JobRunnerRequest> jobRequestQueue = new LinkedBlockingQueue<JobRunnerRequest>();    private static volatile  JobManagerService jobManagerInstnce;    private static Object objSyncLocker = new Object();    private JobManagerService() {    }    private void startupWorker() {        new Thread(new JobWorker()).start();    }    public static JobManagerService getJobManager() {        if (jobManagerInstnce == null) {            synchronized (objSyncLocker) {                if (jobManagerInstnce == null) {                    jobManagerInstnce = new JobManagerService();                    jobManagerInstnce.startupWorker();                }            }        }        return jobManagerInstnce;    }    public void addRequest(JobRunnerRequest request) {        try {            jobRequestQueue.put(request);        }        catch (InterruptedException e) {            LoggerUtil.getJobLogger().warn(e.getMessage(), e);        }    }    public JobRunnerRequest takeRequest() {        try {            return jobRequestQueue.take();        }        catch (InterruptedException e) {            LoggerUtil.getJobLogger().warn(e.getMessage(), e);            return null;        }    }}

import org.springframework.context.ApplicationContext;import org.springframework.context.ApplicationContextAware;public class ApplicationContextProvider implements ApplicationContextAware {    private static volatile ApplicationContext ctx;    public static ApplicationContext getContext() {        return ctx;    }    private static synchronized void setContext(ApplicationContext applicationContext) {        ctx = applicationContext;    }    @Override    public void setApplicationContext(ApplicationContext applicationContext){        setContext(applicationContext);    }}