Database2中的每一行都针对Database1中的相应行进行验证.为了加快进程,我使用Parallel.ForEach循环.

我最初的问题是,这个计数总是与我预想的不同.我后来发现,=和 – =操作不是线程安全的(不是原子的).为了纠正这个问题,我更新了使用Interlocked.Increment的代码来计数变量.这个代码打印一个更接近实际计数的计数,但仍然在每次执行时似乎都有所不同,它没有给出我预期的结果：

Private countObjects As Integer

Private Sub MyMainFunction()
    Dim objects As List(Of MyObject)

    'Query with Dapper,unrelevant to the problem.
    Using connection As New System.Data.sqlClient.sqlConnection("aConnectionString")
        objects = connection.Query("SELECT * FROM Foo") 'Returns around 81000 rows.
    End Using

    Parallel.ForEach(objects,Sub(u) MyParallelFunction(u))

    Console.WriteLine(String.Format("Count : {0}",countObjects)) 'Prints "Count : 80035" or another incorrect count,which seems to differ on each execution of MyMainFunction.
End Sub

Private Sub MyParallelFunction(obj As MyObject)
    Interlocked.Increment(countObjects) 'Breakpoint Hit Count is at around 81300 or another incorrect number when done.

    'Continues executing unrelated code using obj...
End Sub

在使用其他增加线程安全性的方法进行了一些实验之后,我发现将虚拟参考对象上的SyncLock中的增量包装起来会给出预期的结果：

Private countObjects As Integer
Private locker As SomeType

Private Sub MyMainFunction()
    locker = New SomeType()
    Dim objects As List(Of MyObject)

    'Query with Dapper,countObjects)) 'Prints "Count : 81000".
End Sub

Private Sub MyParallelFunction(obj As MyObject)
    SyncLock locker
        countObjects += 1 'Breakpoint Hit Count is 81000 when done.
    End SyncLock

    'Continues executing unrelated code using obj...
End Sub

为什么第一个代码段不按预期工作？最令人困惑的是断点命中计数提供意想不到的结果.

我对Interlocked.Increment或原子操作的理解有缺陷吗？我不想在一个虚拟对象上使用SyncLock,我希望有一个干净的方法.

更新：

>我在任何cpu上运行Debug模式下的示例.
>我在堆栈中使用ThreadPool.SetMaxThreads(60,60),因为我在某个时候查询Access数据库.这可能会造成问题吗？
>可以调用Increment混合Parallel.ForEach循环,强制它在所有任务完成之前退出？

更新2(方法)：

>我的测试执行的代码尽可能接近这里显示的内容,除了对象类型和查询字符串.
>查询总是给出相同数量的结果,我总是验证objects.Count在断点上继续Parallel.ForEach.
>在执行之间更改的唯一代码是Interlocked.Increment替换为SyncLock locker和countObjects = 1.

更新3

我通过在新的控制台应用程序中复制我的代码并替换外部类和代码来创建了一个SSCCE.

这是控制台应用程序的主要方法：

Sub Main()
    Dim oClass1 As New Class1
    oClass1.MyMainFunction()
End Sub

这是Class1的定义：

Imports System.Threading

Public Class Class1

    Public Class Dummy
        Public Sub New()
        End Sub
    End Class

    Public Class MyObject
        Public Property Id As Integer

        Public Sub New(p_Id As Integer)
            Id = p_Id
        End Sub
    End Class

    Public Property countObjects As Integer
    Private locker As Dummy

    Public Sub MyMainFunction()
        locker = New Dummy()
        Dim objects As New List(Of MyObject)

        For i As Integer = 1 To 81000
            objects.Add(New MyObject(i))
        Next

        Parallel.ForEach(objects,Sub(u As MyObject)
                                      MyParallelFunction(u)
                                  End Sub)

        Console.WriteLine(String.Format("Count : {0}",countObjects)) 'Interlock prints an incorrect count,different in each execution. SyncLock prints the correct count.
        Console.ReadLine()
    End Sub

    'Interlocked
    Private Sub MyParallelFunction(ByVal obj As MyObject)
        Interlocked.Increment(countObjects)
    End Sub

    'SyncLock
    'Private Sub MyParallelFunction(ByVal obj As MyObject)
    '    SyncLock locker
    '        countObjects += 1
    '    End SyncLock
    'End Sub

End Class

将MyParallelFunction从Interlocked.Increment切换到SyncLock时,我仍然注意到相同的行为.

Value = <value from Property>
Interlocked.Increment(Value)
<Property> = Value

因此打破了Increment提供的任何线程保证.将其更改为字段. VB将重写任何作为ByRef参数传递的属性到类似于上述的代码.

private static void appendAbbreviated(StringBuilder to,String text,int start,int end,FontMetrics metrics,int maxWidth,boolean replaceLineTerminators) {
  int abbreviationLength = abbreviationLength(text,start,end,metrics,maxWidth,replaceLineTerminators);

  if (!replaceLineTerminators) {
    to.append(text,start + abbreviationLength);
  }
  else {
    CharSequenceSubSequence subSeq = new CharSequenceSubSequence(text,start + abbreviationLength);
    for (Linetokenizer lt = new Linetokenizer(subSeq); !lt.atEnd(); lt.advance()) {
      to.append(subSeq,lt.getoffset(),lt.getoffset() + lt.getLength());
      if (lt.getLineseparatorLength() > 0) {
        to.append(RETURN_SYMBOL);
      }
    }
  }

  if (abbreviationLength != end - start) {
    to.append(ABBREVIATION_SUFFIX);
  }
}

private static int abbreviationLength(String text,boolean replaceSeparators) {
  if (metrics.charWidth('m') * (end - start) <= maxWidth) return end - start;

  int abbrWidth = metrics.charWidth(ABBREVIATION_SUFFIX);
  int abbrLength = 0;

  CharSequenceSubSequence subSeq = new CharSequenceSubSequence(text,end);
  for (Linetokenizer lt = new Linetokenizer(subSeq); !lt.atEnd(); lt.advance()) {
    for (int i = 0; i < lt.getLength(); i++,abbrLength++) {
      abbrWidth += metrics.charWidth(subSeq.charat(lt.getoffset() + i));
      if (abbrWidth >= maxWidth) return abbrLength;
    }
    if (replaceSeparators && lt.getLineseparatorLength() != 0) {
      abbrWidth += metrics.charWidth(RETURN_SYMBOL);
      if (abbrWidth >= maxWidth) return abbrLength;
      abbrLength += lt.getLineseparatorLength();
    }
  }

  return abbrLength;
}

private static void appendAbbreviated(StringBuilder to,lt.getoffset() + lt.getLength());
      if (lt.getLineseparatorLength() > 0) {
        to.append(RETURN_SYMBOL);
      }
    }
  }

  if (abbreviationLength != end - start) {
    to.append(ABBREVIATION_SUFFIX);
  }
}

private static int abbreviationLength(String text,abbrLength++) {
      abbrWidth += metrics.charWidth(subSeq.charat(lt.getoffset() + i));
      if (abbrWidth >= maxWidth) return abbrLength;
    }
    if (replaceSeparators && lt.getLineseparatorLength() != 0) {
      abbrWidth += metrics.charWidth(RETURN_SYMBOL);
      if (abbrWidth >= maxWidth) return abbrLength;
      abbrLength += lt.getLineseparatorLength();
    }
  }

  return abbrLength;
}

private static boolean collectPatternMatches(IndexPattern indexPattern,CharSequence chars,int commentStart,int commentEnd,PsiFile file,TextRange range,Processor<IndexPatternOccurrence> consumer,TIntArrayList matches
                                             ) {
  Pattern pattern = indexPattern.getPattern();
  if (pattern != null) {
    ProgressManager.checkCanceled();

    CharSequence input = new CharSequenceSubSequence(chars,commentStart,commentEnd);
    Matcher matcher = pattern.matcher(input);
    while (true) {
      //long time1 = System.currentTimeMillis();
      boolean found = matcher.find();
      //long time2 = System.currentTimeMillis();
      //System.out.println("scanned text of length " + (lexer.getTokenEnd() - lexer.getTokenStart() + " in " + (time2 - time1) + " ms"));

      if (!found) break;
      int start = matcher.start() + commentStart;
      int end = matcher.end() + commentStart;
      if (start != end) {
        if ((range == null || range.getStartOffset() <= start && end <= range.getEndOffset()) && matches.indexOf(start) == -1) {
          matches.add(start);
          if (!consumer.process(new IndexPatternOccurrenceImpl(file,indexPattern))) {
            return false;
          }
        }
      }

      ProgressManager.checkCanceled();
    }
  }
  return true;
}

@Override
public void start(@NotNull CharSequence buffer,int startOffset,int endOffset,int state) {
  if (myAlreadyInitializedHighlighter) {
    this.buffer = buffer;
    start = startOffset;
    end = endOffset;
  } else {
    myHighlighter.setText(new CharSequenceSubSequence(this.buffer = buffer,start = startOffset,end = endOffset));
  }
  iterator = myHighlighter.createIterator(0);
}

@Override
public void start(CharSequence buffer,end = endOffset));
  }
  iterator = myHighlighter.createIterator(0);
}

private static boolean collectPatternMatches(IndexPattern indexPattern,Processor<IndexPatternOccurrence> consumer) {
  Pattern pattern = indexPattern.getPattern();
  if (pattern != null) {
    ProgressManager.checkCanceled();

    CharSequence input = new CharSequenceSubSequence(chars,commentEnd);
    Matcher matcher = pattern.matcher(input);
    while (true) {
      //long time1 = System.currentTimeMillis();
      boolean found = matcher.find();
      //long time2 = System.currentTimeMillis();
      //System.out.println("scanned text of length " + (lexer.getTokenEnd() - lexer.getTokenStart() + " in " + (time2 - time1) + " ms"));

      if (!found) break;
      int start = matcher.start() + commentStart;
      int end = matcher.end() + commentStart;
      if (start != end) {
        if (range == null || range.getStartOffset() <= start && end <= range.getEndOffset()) {
          if (!consumer.process(new IndexPatternOccurrenceImpl(file,indexPattern))) {
            return false;
          }
        }
      }

      ProgressManager.checkCanceled();
    }
  }
  return true;
}

private static boolean collectPatternMatches(IndexPattern indexPattern,indexPattern))) {
            return false;
          }
        }
      }

      ProgressManager.checkCanceled();
    }
  }
  return true;
}

@Override
public void start(@Nonnull CharSequence buffer,end = endOffset));
  }
  iterator = myHighlighter.createIterator(0);
}

@Override
public PsiFile createFileFromText(FileType fileType,final String fileName,int endOffset) {
  LOG.assertTrue(!fileType.isBinary());
  final CharSequence text =
    startOffset == 0 && endOffset == chars.length() ? chars : new CharSequenceSubSequence(chars,startOffset,endOffset);
  return createFileFromText(fileName,fileType,text);
}

@Override
public void start(CharSequence buffer,end = endOffset));
  }
  iterator = myHighlighter.createIterator(0);
}

@Override
public PsiFile createFileFromText(FileType fileType,int endOffset) {
  LOG.assertTrue(!fileType.isBinary());
  final CharSequence text = startOffset == 0 && endOffset == chars.length()?chars:new CharSequenceSubSequence(chars,text);
}

@NotNull
@Override
public CharSequence intern(@NotNull final CharSequence baseText,final int startOffset,final int endOffset) {
  if (endOffset - startOffset == baseText.length()) return baseText;
  return intern(new CharSequenceSubSequence(baseText,endOffset));
}

@Override
public PsiFile createFileFromText(FileType fileType,text);
}