在flink连接流中实现过程函数时面临竞争条件。我正在吃 Cache Map 在两个职能部门之间共享 processElement1 & processElement2 由两个不同的线程并行调用。 Streams1 --->（发送报价数据） Streams2 --->（发送lms（忠诚度管理系统数据））

connect=Streams1.connect(Streams2);

connect.process(new TriggerStream);

在 TriggerStream Class 我正在使用唯一id存储数据： MemberId 作为 unique Key 存储和 lookup data 在缓存中。当数据流入时，我没有得到任何结果

class LRUConcurrentCache<K,V>{
    private final Map<K,V> cache;
    private final int maxEntries;
    public LRUConcurrentCache(final int maxEntries) {
        this.cache = new LinkedHashMap<K,V>(maxEntries, 0.75F, true) {
            private static final long serialVersionUID = -1236481390177598762L;
            @Override
            protected boolean removeEldestEntry(Map.Entry<K,V> eldest){
                return size() > maxEntries;
            }
        };
    }
    //Why we cant lock on the key
    public void put(K key, V value) {
        synchronized(key) {
            cache.put(key, value);
        }
    }

    //get methode
    public V get(K key) {
        synchronized(key) {
            return cache.get(key);
        }
    }

public class TriggerStream extends CoProcessFunction<IOffer, LMSData, String> {

    private static final long serialVersionUID = 1L;
    LRUCache cache; 
    private String offerNode;
    String updatedValue, retrivedValue;
    Subscriber subscriber;

    TriggerStream(){
        this.cache== new LRUCache(10);
    }

@Override
    public void processElement1(IOffer offer) throws Exception {
        try {
            ObjectMapper mapper = new ObjectMapper();
            mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
            mapper.enableDefaultTyping();
            // TODO Auto-generated method stub
            IOffer latestOffer = offer;

            //Check the subscriber is there or not

            retrivedValue = cache.get(latestOffer.getMemberId().toString());
            if ((retrivedValue == null)) {
                //Subscriber is the class that is used and converted as Json String & then store into map
                Subscriber subscriber = new Subscriber();
                subscriber.setMemberId(latestOffer.getMemberId());
                ArrayList<IOffer> offerList = new ArrayList<IOffer>();
                offerList.add(latestOffer);
                subscriber.setOffers(offerList);
                updatedValue = mapper.writeValueAsString(subscriber);
                cache.set(subscriber.getMemberId().toString(), updatedValue);
            } else {
                Subscriber subscriber = mapper.readValue(retrivedValue, Subscriber.class);
                List<IOffer> offers = subscriber.getOffers();
                offers.add(latestOffer);
                updatedValue= mapper.writeValueAsString(subscriber);
                cache.set(subscriber.getMemberId().toString(), subscriberUpdatedValue);
            }
        } catch (Exception pb) {
            applicationlogger.error("Exception in Offer Loading:"+pb);
            applicationlogger.debug("*************************FINISHED OFFER LOADING*******************************");
        }
        applicationlogger.debug("*************************FINISHED OFFER LOADING*******************************");

    }

@Override
    public void processElement2(LMSData lms) throws Exception {
        try {
            ObjectMapper mapper = new ObjectMapper();
            mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
            mapper.enableDefaultTyping();
            // TODO Auto-generated method stub

            //Check the subscriber is there or not

            retrivedValue = cache.get(lms.getMemberId().toString());
            if(retrivedValue !=null){
                Subscriber subscriber = mapper.readValue(retrivedValue, Subscriber.class);
                //do some calculations 
                String updatedValue = mapper.writeValueAsString(subscriber);
                //Update value
                cache.set(subscriber.getMemberId().toString(), updatedValue);
            }

        } catch (Exception pb) {
            applicationlogger.error("Exception in Offer Loading:"+pb);
            applicationlogger.debug("*************************FINISHED OFFER LOADING*******************************");
        }
        applicationlogger.debug("*************************FINISHED OFFER LOADING*******************************");

    }

}