请看下面的方法：
public void add(final List<ReportingSTG> message) {
if(stopRequested.get()) {
synchronized (this) {
if(stopRequested.get()) {
retryQueue.put(message);
}
}
}
messages.add(message);
if(messages.size() >= batchSize && waitingThreads.get() == 0) {
synchronized (this) {
if(messages.size() >= batchSize && waitingThreads.get() == 0) {
final List<List<ReportingSTG>> clone = new ArrayList<List<ReportingSTG>>(messages);
messages.clear();
if(processors.size()>=numOfProcessors) {
waitingThreads.incrementAndGet();
waitForProcessor();
waitingThreads.decrementAndGet();
}
startProcessor(clone);
}
}
}
}
尤其这2行：
1: final List<List<ReportingSTG>> clone = new ArrayList<List<ReportingSTG>>(messages);
2: messages.clear();
如果线程A进入同步块和当前对象上获取锁，确实这意味着该对象的实例属性的状态不能由同步块外其他线程被改变（而线程A是在同步块）？
例如，线程A执行第1行 - >线程B输入的方法和添加新的列表条目（messages.add（消息）） - >线程执行第2行 - >条目怎样通过线程B与其它项添加去除（一起）。 这情景可能吗？ 或者线程B将等待锁被线程A发布，然后才将删除列表条目
消息是一个非静态synchronizedList
UPD：更新方法，可能的解决方案：
public void add(final List<ReportingSTG> message) {
if(stopRequested.get()) {
synchronized (this) {
if(stopRequested.get()) {
retryQueue.put(message);
}
}
}
while (addLock.get()){
try {
Thread.sleep(1);
} catch (InterruptedException e) {}
}
messages.add(message);
if(messages.size() >= batchSize && waitingThreads.get() == 0) {
synchronized (this) {
if(messages.size() >= batchSize && waitingThreads.get() == 0) {
addLock.set(true);
final List<List<ReportingSTG>> clone = new ArrayList<List<ReportingSTG>>(messages);
messages.clear();
addLock.set(false);
if(processors.size()>=numOfProcessors) {
waitingThreads.incrementAndGet();
waitForProcessor();
waitingThreads.decrementAndGet();
}
startProcessor(clone);
}
}
}
}
addLock - 的AtomicBoolean，默认为false
所描述的情况是可能的。 也就是说，你可能会失去消息。
该synchronized关键字确保您永远不会有2个线程运行synchronized同时部分。 它并不防止修改由在内部操纵对象的另一个线程 synchronized块（只要该其他线程访问它们）。
这是一个可能的解决方案，因为它同步添加和清晰。
private Object lock = new Object();
public void add(final List<ReportingSTG> message) {
if(stopRequested.get()) {
synchronized (this) {
if(stopRequested.get()) {
retryQueue.put(message);
}
}
}
synchronized(lock){
messages.add(message);
if(messages.size() >= batchSize && waitingThreads.get() == 0) {
final List<List<ReportingSTG>> clone = new ArrayList<List<ReportingSTG>>(messages);
messages.clear();
if(processors.size()>=numOfProcessors) {
waitingThreads.incrementAndGet();
waitForProcessor();
waitingThreads.decrementAndGet();
}
startProcessor(clone);
}
}
}
我把一个DoubleBufferedList最近的类。 也许使用将避免您的问题彻底。 正如它的名字所暗示它实现了双缓冲算法，但对列表。
这个类让你有许多生产者线程和许多消费者线程。 每个生产者线程可以添加到当前列表。 每个消费者线程获取用于处理当前整个列表。
这也使用没有锁，只是原子能所以应该有效地运行。
请注意，这在很大程度上是测试代码。 您可以在之后删除一切// TESTING发表评论，但你可以找到安慰测试的严谨性。
public class DoubleBufferedList<T> {
// Atomic reference so I can atomically swap it through.
// Mark = true means I am adding to it so unavailable for iteration.
private AtomicMarkableReference<List<T>> list = new AtomicMarkableReference<>(newList(), false);
// Factory method to create a new list - may be best to abstract this.
protected List<T> newList() {
return new ArrayList<>();
}
// Get and replace the current list.
public List<T> get() {
// Atomically grab and replace the list with an empty one.
List<T> empty = newList();
List<T> it;
// Replace an unmarked list with an empty one.
if (!list.compareAndSet(it = list.getReference(), empty, false, false)) {
// Failed to replace!
// It is probably marked as being appended to but may have been replaced by another thread.
// Return empty and come back again soon.
return Collections.EMPTY_LIST;
}
// Successfull replaced an unmarked list with an empty list!
return it;
}
// Grab and lock the list in preparation for append.
private List<T> grab() {
List<T> it;
// We cannot fail so spin on get and mark.
while (!list.compareAndSet(it = list.getReference(), it, false, true)) {
// Spin on mark.
}
return it;
}
// Release the list.
private void release(List<T> it) {
// Unmark it. Should never fail because once marked it will not be replaced.
if (!list.attemptMark(it, false)) {
throw new IllegalMonitorStateException("it changed while we were adding to it!");
}
}
// Add an entry to the list.
public void add(T entry) {
List<T> it = grab();
try {
// Successfully marked! Add my new entry.
it.add(entry);
} finally {
// Always release after a grab.
release(it);
}
}
// Add many entries to the list.
public void add(List<T> entries) {
List<T> it = grab();
try {
// Successfully marked! Add my new entries.
it.addAll(entries);
} finally {
// Always release after a grab.
release(it);
}
}
// Add a number of entries.
public void add(T... entries) {
// Make a list of them.
add(Arrays.asList(entries));
}
// TESTING.
// How many testers to run.
static final int N = 10;
// The next one we're waiting for.
static final AtomicInteger[] seen = new AtomicInteger[N];
// The ones that arrived out of order.
static final Set<Widget>[] queued = new ConcurrentSkipListSet[N];
static {
// Populate the arrays.
for (int i = 0; i < N; i++) {
seen[i] = new AtomicInteger();
queued[i] = new ConcurrentSkipListSet();
}
}
// Thing that is produced and consumed.
private static class Widget implements Comparable<Widget> {
// Who produced it.
public final int producer;
// Its sequence number.
public final int sequence;
public Widget(int producer, int sequence) {
this.producer = producer;
this.sequence = sequence;
}
@Override
public String toString() {
return producer + "\t" + sequence;
}
@Override
public int compareTo(Widget o) {
// Sort on producer
int diff = Integer.compare(producer, o.producer);
if (diff == 0) {
// And then sequence
diff = Integer.compare(sequence, o.sequence);
}
return diff;
}
}
// Produces Widgets and feeds them to the supplied DoubleBufferedList.
private static class TestProducer implements Runnable {
// The list to feed.
final DoubleBufferedList<Widget> list;
// My ID
final int id;
// The sequence we're at
int sequence = 0;
// Set this at true to stop me.
public volatile boolean stop = false;
public TestProducer(DoubleBufferedList<Widget> list, int id) {
this.list = list;
this.id = id;
}
@Override
public void run() {
// Just pump the list.
while (!stop) {
list.add(new Widget(id, sequence++));
}
}
}
// Consumes Widgets from the suplied DoubleBufferedList
private static class TestConsumer implements Runnable {
// The list to bleed.
final DoubleBufferedList<Widget> list;
// My ID
final int id;
// Set this at true to stop me.
public volatile boolean stop = false;
public TestConsumer(DoubleBufferedList<Widget> list, int id) {
this.list = list;
this.id = id;
}
@Override
public void run() {
// The list I am working on.
List<Widget> l = list.get();
// Stop when stop == true && list is empty
while (!(stop && l.isEmpty())) {
// Record all items in list as arrived.
arrived(l);
// Grab another list.
l = list.get();
}
}
private void arrived(List<Widget> l) {
for (Widget w : l) {
// Mark each one as arrived.
arrived(w);
}
}
// A Widget has arrived.
private static void arrived(Widget w) {
// Which one is it?
AtomicInteger n = seen[w.producer];
// Don't allow multi-access to the same producer data or we'll end up confused.
synchronized (n) {
// Is it the next to be seen?
if (n.compareAndSet(w.sequence, w.sequence + 1)) {
// It was the one we were waiting for! See if any of the ones in the queue can now be consumed.
for (Iterator<Widget> i = queued[w.producer].iterator(); i.hasNext();) {
Widget it = i.next();
// Is it in sequence?
if (n.compareAndSet(it.sequence, it.sequence + 1)) {
// Done with that one too now!
i.remove();
} else {
// Found a gap! Stop now.
break;
}
}
} else {
// Out of sequence - Queue it.
queued[w.producer].add(w);
}
}
}
}
// Main tester
public static void main(String args[]) {
try {
System.out.println("DoubleBufferedList:Test");
// Create my test buffer.
DoubleBufferedList<Widget> list = new DoubleBufferedList<>();
// All threads running - Producers then Consumers.
List<Thread> running = new LinkedList<>();
// Start some producer tests.
List<TestProducer> producers = new ArrayList<>();
for (int i = 0; i < N; i++) {
TestProducer producer = new TestProducer(list, i);
Thread t = new Thread(producer);
t.setName("Producer " + i);
t.start();
producers.add(producer);
running.add(t);
}
// Start the same number of consumers.
List<TestConsumer> consumers = new ArrayList<>();
for (int i = 0; i < N; i++) {
TestConsumer consumer = new TestConsumer(list, i);
Thread t = new Thread(consumer);
t.setName("Consumer " + i);
t.start();
consumers.add(consumer);
running.add(t);
}
// Wait for a while.
Thread.sleep(5000);
// Close down all.
for (TestProducer p : producers) {
p.stop = true;
}
for (TestConsumer c : consumers) {
c.stop = true;
}
// Wait for all to stop.
for (Thread t : running) {
System.out.println("Joining " + t.getName());
t.join();
}
// What results did we get?
for (int i = 0; i < N; i++) {
// How far did the producer get?
int gotTo = producers.get(i).sequence;
// The consumer's state
int seenTo = seen[i].get();
Set<Widget> queue = queued[i];
if (seenTo == gotTo && queue.isEmpty()) {
System.out.println("Producer " + i + " ok.");
} else {
// Different set consumed as produced!
System.out.println("Producer " + i + " Failed: gotTo=" + gotTo + " seenTo=" + seenTo + " queued=" + queue);
}
}
} catch (InterruptedException ex) {
ex.printStackTrace();
}
}
}
