问题陈述

我有一个JMS监听器作为一个线程来监听一个主题.一旦消息进入,我就会生成一个新的Thread来处理有界消息.因此,对于每个传入的消息,我产生一个新的线程.

我有一个场景,当重复消息按顺序注入时,也会处理重复消息.我需要防止这个被处理.我尝试使用ConcurrentHashMap来保存进程时间,一旦Thread生成就将其添加到条目中,并在Thread完成执行后立即将其从地图中删除.但是当我尝试使用同时以同时方式传递相同的场景时,它没有帮助.

在您深入了解实际代码库之前,我的问题的概要

onMessage(){

processIncomingMessage(){

ExecutorService executorService = Executors.newFixedThreadPool(1000);

//Map is used to make an entry before i spawn a new thread to process incoming message

//Map contains "Key as the incoming message" and "value as boolean"

//check map for duplicate check

//The below check is failing and allowing duplicate messages to be processed in parallel

if(entryisPresentInMap){

//return doing nothing

}else{

//spawn a new thread for each incoming message

//also ensure a duplicate message being processed when it in process by an active thread

executorService.execute(new Runnable() {

@Override

public void run() {

try {

//actuall business logic

}finally{

//remove entry from the map so after processing is done with the message

}

}

}

}

用于模拟场景的独立示例

public class DuplicateCheck {

private static Map duplicateCheckMap =

new ConcurrentHashMap(1000);

private static String name=null;

private static String[] nameArray = new String[20];

public static void processMessage(String message){

System.out.println("Processed message =" +message);

}

public static void main(String args[]){

nameArray[0] = "Peter";

nameArray[1] = "Peter";

nameArray[2] = "Adam";

for(int i=0;i<=nameArray.length;i++){

name=nameArray[i];

if(duplicateCheckMap.get(name)!=null && duplicateCheckMap.get(name)){

System.out.println("Thread detected for processing your name ="+name);

return;

}

addNameIntoMap(name);

new Thread(new Runnable() {

@Override

public void run() {

try {

processMessage(name);

} catch (Exception e) {

System.out.println(e.getMessage());

} finally {

freeNameFromMap(name);

}

}

}).start();

}

}

private static synchronized void addNameIntoMap(String name) {

if (name != null) {

duplicateCheckMap.put(name, true);

System.out.println("Thread processing the "+name+" is added to the status map");

}

}

private static synchronized void freeNameFromMap(String name) {

if (name != null) {

duplicateCheckMap.remove(name);

System.out.println("Thread processing the "+name+" is released from the status map");

}

}

代码片段如下

public void processControlMessage(final Message message) {

RDPWorkflowControlMessage rdpWorkflowControlMessage= unmarshallControlMessage(message);

final String workflowName = rdpWorkflowControlMessage.getWorkflowName();

final String controlMessageEvent=rdpWorkflowControlMessage.getControlMessage().value();

if(controlMessageStateMap.get(workflowName)!=null && controlMessageStateMap.get(workflowName)){

log.info("Cache cleanup for the workflow :"+workflowName+" is already in progress");

return;

}else {

log.info("doing nothing");

}

Semaphore controlMessageLock = new Semaphore(1);

try{

controlMessageLock.acquire();

synchronized(this){

new Thread(new Runnable(){

@Override

public void run() {

try {

lock.lock();

log.info("Processing Workflow Control Message for the workflow :"+workflowName);

if (message instanceof TextMessage) {

if ("REFRESH".equalsIgnoreCase(controlMessageEvent)) {

clearControlMessageBuffer();

enableControlMessageStatus(workflowName);

List matchingValues=new ArrayList();

matchingValues.add(workflowName);

ConcreteSetDAO tasksSetDAO=taskEventListener.getConcreteSetDAO();

ConcreteSetDAO workflowSetDAO=workflowEventListener.getConcreteSetDAO();

tasksSetDAO.deleteMatchingRecords(matchingValues);

workflowSetDAO.deleteMatchingRecords(matchingValues);

fetchNewWorkflowItems();

addShutdownHook(workflowName);

}

}

} catch (Exception e) {

log.error("Error extracting item of type RDPWorkflowControlMessage from message "

+ message);

} finally {

disableControlMessageStatus(workflowName);

lock.unlock();

}

}

}).start();

}

} catch (InterruptedException ie) {

log.info("Interrupted Exception during control message lock acquisition"+ie);

}finally{

controlMessageLock.release();

}

}

private void addShutdownHook(final String workflowName) {

Runtime.getRuntime().addShutdownHook(new Thread() {

public void run() {

disableControlMessageStatus(workflowName);

}

});

log.info("Shut Down Hook Attached for the thread processing the workflow :"+workflowName);

}

private RDPWorkflowControlMessage unmarshallControlMessage(Message message) {

RDPWorkflowControlMessage rdpWorkflowControlMessage = null;

try {

TextMessage textMessage = (TextMessage) message;

rdpWorkflowControlMessage = marshaller.unmarshalItem(textMessage.getText(), RDPWorkflowControlMessage.class);

} catch (Exception e) {

log.error("Error extracting item of type RDPWorkflowTask from message "

+ message);

}

return rdpWorkflowControlMessage;

}

private void fetchNewWorkflowItems() {

initSSL();

List allTasks=initAllTasks();

taskEventListener.addRDPWorkflowTasks(allTasks);

workflowEventListener.updateWorkflowStatus(allTasks);

}

private void clearControlMessageBuffer() {

taskEventListener.getRecordsForUpdate().clear();

workflowEventListener.getRecordsForUpdate().clear();

}

private synchronized void enableControlMessageStatus(String workflowName) {

if (workflowName != null) {

controlMessageStateMap.put(workflowName, true);

log.info("Thread processing the "+workflowName+" is added to the status map");

}

}

private synchronized void disableControlMessageStatus(String workflowName) {

if (workflowName != null) {

controlMessageStateMap.remove(workflowName);

log.info("Thread processing the "+workflowName+" is released from the status map");

}

}

我修改了我的代码以包含下面提供的建议,但仍然无效

public void processControlMessage(final Message message) {

ExecutorService executorService = Executors.newFixedThreadPool(1000);

try{

lock.lock();

RDPWorkflowControlMessage rdpWorkflowControlMessage= unmarshallControlMessage(message);

final String workflowName = rdpWorkflowControlMessage.getWorkflowName();

final String controlMessageEvent=rdpWorkflowControlMessage.getControlMessage().value();

if(controlMessageStateMap.get(workflowName)!=null && controlMessageStateMap.get(workflowName)){

log.info("Cache cleanup for the workflow :"+workflowName+" is already in progress");

return;

}else {

log.info("doing nothing");

}

enableControlMessageStatus(workflowName);

executorService.execute(new Runnable() {

@Override

public void run() {

try {

//actual code

fetchNewWorkflowItems();

addShutdownHook(workflowName);

}

}

} catch (Exception e) {

log.error("Error extracting item of type RDPWorkflowControlMessage from message "

+ message);

} finally {

disableControlMessageStatus(workflowName);

}

}

});

} finally {

executorService.shutdown();

lock.unlock();

}

}

private void addShutdownHook(final String workflowName) {

Runtime.getRuntime().addShutdownHook(new Thread() {

public void run() {

disableControlMessageStatus(workflowName);

}

});

log.info("Shut Down Hook Attached for the thread processing the workflow :"+workflowName);

}

private synchronized void enableControlMessageStatus(String workflowName) {

if (workflowName != null) {

controlMessageStateMap.put(workflowName, true);

log.info("Thread processing the "+workflowName+" is added to the status map");

}

}

private synchronized void disableControlMessageStatus(String workflowName) {

if (workflowName != null) {

controlMessageStateMap.remove(workflowName);

log.info("Thread processing the "+workflowName+" is released from the status map");

}

}

解决方法:

这是您应该如何向地图添加值.这种双重检查确保在任何特定时刻只有一个线程向地图添加值,然后您可以控制访问.之后删除所有锁定逻辑.它是如此简单

public void processControlMessage(final String workflowName) {

if(!tryAddingMessageInProcessingMap(workflowName)){

Thread.sleep(1000); // sleep 1 sec and try again

processControlMessage(workflowName);

return ;

}

System.out.println(workflowName);

try{

// your code goes here

} finally{

controlMessageStateMap.remove(workflowName);

}

}

private boolean tryAddingMessageInProcessingMap(final String workflowName) {

if(controlMessageStateMap .get(workflowName)==null){

synchronized (this) {

if(controlMessageStateMap .get(workflowName)==null){

controlMessageStateMap.put(workflowName, true);

return true;

}

}

}

return false;

}

标签：java,multithreading,jms-topic

来源： https://codeday.me/bug/20190706/1397427.html