아래로
코드 조각은 QueueMessageListener가 비동기 적으로 실행되고 있습니다,이 작업을 수행하지만 설계에 문제가 왜, 아래의주의 사항 1 5 참조 알고 그것을 바꿀 수있다 settext.setTextmessage((TextMessage) message); 또 다른 소비자는 끝난 TextMessage를 검색하고이 어쩌면 V2가 더하지만 어쩌면 org.springframework.jms.listener.DefaultMessageListenerContainer 최고의 솔루션 사용을 위해, 큐에 추가하기 전에 :
public static void main(String[] args) throws InterruptedException, JMSException {
//Create a producer
// 1- settext.getTextmessage() == null i suppose at this level, see 2- point
Thread producer = new Thread(new Producer(queue,settext));
producer.start();
//Create a Consumer with coresize 4 and Max size 10
final ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 10, 100, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(), new ThreadPoolExecutor.CallerRunsPolicy());
executor.allowCoreThreadTimeOut(true);
// 3- you start consumers go to 4, note that you will only consume count messages !!
for (int i = 0; i <count; i++) {
executor.execute(new Consumer(queue));
}
**//INITIALIZE ACTIVEMQ CONFIGURATION HERE**
consumer.setMessageListener(new QueueMessageListener());
executor.shutdown();
}
private static class QueueMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
//Setting the text message to a setter which takes TextMessage as arg
settext.setTextmessage((TextMessage) message);
// at this point message is considered as delivered if sessionAcknowledgeModeName is AUTO_ACKNOWLEDGE and maybe lost if asynchronous treatment fails
}
}
}
//Problem here unable to produce
class Producer implements Runnable {
ConcurrentLinkedQueue<TextMessage> queue;
Settext settext;
Producer(ConcurrentLinkedQueue<TextMessage> queue2, Settext settext){
this.queue = queue2;
this.settext=settext;
}
public void run() {
System.out.println("Producer Started");
try {
// 2- settext.getTextmessage() == null if block is not executed and thread will sleep and finish
// you have to add this
while (this.settext.getTextmessage() == null) {
try {
Thread.currentThread().sleep(500);
} catch (Exception ex) {
ex.printStackTrace();
}
}
if(this.settext.getTextmessage()!=null)
{
//Add to ConcurrentLinkedQueue
queue.add(this.settext.getTextmessage());
}
//}
} catch (Exception ex) {
ex.printStackTrace();
}
}
}
//Problem here unable to consume
class Consumer implements Runnable {
ConcurrentLinkedQueue<TextMessage> queue;
public Consumer(ConcurrentLinkedQueue<TextMessage> queue2) {
this.queue = queue2;
}
public void run() {
TextMessage str;
System.out.println("Consumer Started");
// 4- queue.poll() == null at this level, while loop finished, thread will sleep and finish
// you have to add this
while ((str = queue.poll()) == null) {
try {
Thread.currentThread().sleep(500);
} catch (Exception ex) {
ex.printStackTrace();
}
}
System.out.println("Removed: " + str);
//}
}
V2 :
,451,515,
public static void main(String[] args) throws InterruptedException, JMSException {
//Create a Consumer with coresize 4 and Max size 10
final ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 10, 100, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(), new ThreadPoolExecutor.CallerRunsPolicy());
executor.allowCoreThreadTimeOut(true);
for (int i = 0; i <count; i++) {
executor.execute(new Consumer(queue));
}
**//INITIALIZE ACTIVEMQ CONFIGURATION HERE**
consumer.setMessageListener(new QueueMessageListener());
executor.shutdown();
}
private static class QueueMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
queue.add((TextMessage) message);
}
}
}
//Problem here unable to consume
class Consumer implements Runnable {
ConcurrentLinkedQueue<TextMessage> queue;
public Consumer(ConcurrentLinkedQueue<TextMessage> queue2) {
this.queue = queue2;
}
public void run() {
TextMessage str;
System.out.println("Consumer Started");
while ((str = queue.poll()) == null) {
try {
Thread.currentThread().sleep(500);
} catch (Exception ex) {
ex.printStackTrace();
}
}
System.out.println("Removed: " + str);
//}
}
V3 :
public static void main(String[] args) throws InterruptedException, JMSException {
//Create a Consumer with coresize 4 and Max size 10
final ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 10, 100, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>(), new ThreadPoolExecutor.CallerRunsPolicy());
executor.allowCoreThreadTimeOut(true);
**//INITIALIZE ACTIVEMQ CONFIGURATION HERE**
consumer.setMessageListener(new QueueMessageListener());
executor.shutdown();
}
private static class QueueMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
executor.execute(new Consumer((TextMessage) message));
}
}
}
//Problem here unable to consume
class Consumer implements Runnable {
TextMessage textMessage;
public Consumer(TextMessage textMessage) {
this.textMessage = textMessage;
}
public void run() {
System.out.println("Removed: " + str);
}
}
V4 :
public static void main(String[] args) throws InterruptedException, JMSException {
new Consumer(queue).start();
**//INITIALIZE ACTIVEMQ CONFIGURATION HERE**
consumer.setMessageListener(new QueueMessageListener());
executor.shutdown();
}
private static class QueueMessageListener implements MessageListener {
@Override
public void onMessage(Message message) {
queue.add((TextMessage) message);
}
}
//Problem here unable to consume
class Consumer implements Runnable {
ConcurrentLinkedQueue<TextMessage> queue;
public Consumer(ConcurrentLinkedQueue<TextMessage> queue2) {
this.queue = queue2;
}
public void run() {
TextMessage str;
System.out.println("Consumer Started");
while (true) {
try {
Thread.currentThread().sleep(500);
} catch (Exception ex) {
}
while ((str = queue.poll()) == null) {
try {
Thread.currentThread().sleep(500);
} catch (Exception ex) {
}
}
System.out.println("Removed: " + str);
}
}
}
스피 ActiveMQ를 producer.I에서 100 개의 메시지를 보내는 코드의 V2를 시도하지만 난 ("제거"에서 System.out.println + str); 10 번만 인쇄합니다. 즉 큐에서 10 개의 요소 만 제거합니다. 잘못된 경우 –
이 경우 변수 개수 == 10을 의미합니까? –
어쩌면 V3가 더 유연합니다 –