Scala演员效率低下问题

Question

让我先说我是斯卡拉的新手; 但是,我发现基于Actor的并发模型很有趣,我试着给它一个相对简单的应用程序.我遇到的问题是,虽然我能够让应用程序工作,但结果远远低于基于Java的等效解决方案的效率(就实时,CPU时间和内存使用而言).使用从ArrayBlockingQueue中提取消息的线程.我想明白为什么.我怀疑这可能是我缺乏Scala知识,而且我造成了所有低效率,但经过多次尝试重新编写应用程序但没有成功,我决定联系社区寻求帮助.

我的问题是:我有一个gzip压缩文件,其格式为:

SomeID comma_separated_list_of_values

例如:

1234 12,45,82

我想解析每一行,并得到逗号分隔列表中每个值的出现次数的总体计数.

此文件可能非常大(几GB压缩),但每个文件的唯一值数量非常小(最多500个).我认为这是一个尝试编写基于Actor的并发Scala应用程序的好机会.我的解决方案涉及一个主驱动程序,它创建一个解析器Actors池.然后主驱动程序从stdin读取行,将行传递给一个解析行的Actor并保持值的本地计数.当主驱动程序读取最后一行时,它会向每个actor传递一条消息,指示已读取所有行.当演员收到"完成"消息时,他们将计数传递给聚合器,该聚合器对所有演员的计数进行求和.一旦聚合了所有解析器的计数,主驱动程序就会打印出统计信息.

问题:我遇到的主要问题是这个应用程序的低效率.与使用线程和ArrayBlockingQueue的"等效"Java应用程序相比,它使用了更多的CPU和更多的内存.为了正确看待这一点,我收集了一些1000万行测试输入文件的统计数据:

Scala 1 Actor(解析器):

    real    9m22.297s
    user    235m31.070s
    sys     21m51.420s

Java 1线程(解析器):

    real    1m48.275s
    user    1m58.630s
    sys     0m33.540s

Scala 5演员:

    real    2m25.267s
    user    63m0.730s
    sys     3m17.950s

Java 5主题:

    real    0m24.961s
    user    1m52.650s
    sys     0m20.920s

此外,top报告称Scala应用程序的驻留内存大小约为10倍.所以我们谈论的是更多的CPU和内存量级,以达到数量级更差的性能,而我无法弄清楚导致这种情况的原因.这是一个GC问题,还是我以某种方式创建了比我意识到的更多的对象副本？

其他可能重要或可能不重要的细节:

• scala应用程序由Java类包装,因此我可以提供一个自包含的可执行JAR文件(我在每台可能要运行此应用程序的计算机上都没有Scala jar).
• 正在调用该应用程序如下:gunzip -c gzFilename | java -jar StatParser.jar

这是代码:

主要司机:

import scala.actors.Actor._
import scala.collection.{ immutable, mutable }
import scala.io.Source

class StatCollector (numParsers : Int ) {
    private val parsers = new mutable.ArrayBuffer[StatParser]()
    private val aggregator = new StatAggregator()

    def generateParsers {
        for ( i <- 1 to numParsers ) {
            val parser = new StatParser( i, aggregator )
            parser.start
            parsers += parser
        }
    }


    def readStdin {
        var nextParserIdx = 0
        var lineNo = 1
        for ( line <- Source.stdin.getLines() ) {
            parsers( nextParserIdx ) ! line
            nextParserIdx += 1
            if ( nextParserIdx >= numParsers ) {
                nextParserIdx = 0
            }
            lineNo += 1
        }
    }

    def informParsers {
        for ( parser <- parsers ) {
            parser ! true
        }
    }

    def printCounts {
        val countMap = aggregator.getCounts()
        println( "ID,Count" )
        /*
        for ( key <- countMap.keySet ) {
            println( key + "," + countMap.getOrElse( key, 0 ) )
            //println( "Campaign '" + key + "': " + countMap.getOrElse( key, 0 ) )
        }
        */
        countMap.toList.sorted foreach {
            case (key, value) =>
                println( key + "," + value )
        }
    }

    def processFromStdIn {
        aggregator.start

        generateParsers

        readStdin
        process
    }

    def process {

        informParsers

        var completedParserCount = aggregator.getNumParsersAggregated
        while ( completedParserCount < numParsers ) {
            Thread.sleep( 250 )
            completedParserCount = aggregator.getNumParsersAggregated
        }

        printCounts
    }
}

解析演员:

import scala.actors.Actor
import collection.mutable.HashMap
import scala.util.matching

class StatParser( val id: Int, val aggregator: StatAggregator ) extends Actor {

    private var countMap = new HashMap[String, Int]()
    private val sep1 = "\t"
    private val sep2 = ","


    def getCounts(): HashMap[String, Int] = {
        return countMap
    }

    def act() {
        loop {
            react {
                case line: String =>
                    {
                        val idx = line.indexOf( sep1 )
                        var currentCount = 0
                        if ( idx > 0 ) {
                            val tokens = line.substring( idx + 1 ).split( sep2 )
                            for ( token <- tokens ) {
                                if ( !token.equals( "" ) ) {
                                    currentCount = countMap.getOrElse( token, 0 )
                                    countMap( token ) = ( 1 + currentCount )
                                }
                            }

                        }
                    }
                case doneProcessing: Boolean =>
                    {
                        if ( doneProcessing ) {
                            // Send my stats to Aggregator
                            aggregator ! this
                        }
                    }
            }
        }
    }
}

聚合器角色:

import scala.actors.Actor
import collection.mutable.HashMap

class StatAggregator extends Actor {
    private var countMap = new HashMap[String, Int]()
    private var parsersAggregated = 0

    def act() {
        loop {
            react {
                case parser: StatParser =>
                    {
                        val cm = parser.getCounts()
                        for ( key <- cm.keySet ) {
                            val currentCount = countMap.getOrElse( key, 0 )
                            val incAmt = cm.getOrElse( key, 0 )
                            countMap( key ) = ( currentCount + incAmt )
                        }
                        parsersAggregated += 1
                    }
            }
        }
    }

    def getNumParsersAggregated: Int = {
        return parsersAggregated
    }

    def getCounts(): HashMap[String, Int] = {
        return countMap
    }
}

在理解这里发生的事情时可以提供的任何帮助将不胜感激.

提前致谢!

----编辑---

由于许多人回应并要求提供Java代码,因此这里是我为比较目的而创建的简单Java应用程序.我意识到这不是很好的Java代码,但是当我看到Scala应用程序的性能时,我只是快速地看到一些基于Java线程的实现如何作为基线执行:

解析线程:

import java.util.Hashtable;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;

public class JStatParser extends Thread
{
    private ArrayBlockingQueue<String> queue;
    private Map<String, Integer> countMap;
    private boolean done;

    public JStatParser( ArrayBlockingQueue<String> q )
    {
        super( );
        queue = q;
        countMap = new Hashtable<String, Integer>( );
        done = false;
    }

    public Map<String, Integer> getCountMap( )
    {
        return countMap;
    }

    public void alldone( )
    {
        done = true;
    }

    @Override
    public void run( )
    {
        String line = null;
        while( !done || queue.size( ) > 0 )
        {
            try
            {
                // line = queue.take( );
                line = queue.poll( 100, TimeUnit.MILLISECONDS );
                if( line != null )
                {
                    int idx = line.indexOf( "\t" ) + 1;
                    for( String token : line.substring( idx ).split( "," ) )
                    {
                        if( !token.equals( "" ) )
                        {
                            if( countMap.containsKey( token ) )
                            {
                                Integer currentCount = countMap.get( token );
                                currentCount++;
                                countMap.put( token, currentCount );
                            }
                            else
                            {
                                countMap.put( token, new Integer( 1 ) );
                            }
                        }
                    }
                }
            }
            catch( InterruptedException e )
            {
                // TODO Auto-generated catch block
                System.err.println( "Failed to get something off the queue: "
                        + e.getMessage( ) );
                e.printStackTrace( );
            }
        }
    }
}

司机:

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.TreeSet;
import java.util.concurrent.ArrayBlockingQueue;

public class JPS
{
    public static void main( String[] args )
    {
        if( args.length <= 0 || args.length > 2 || args[0].equals( "-?" ) )
        {
            System.err.println( "Usage: JPS [filename]" );
            System.exit( -1 );
        }

        int numParsers = Integer.parseInt( args[0] );
        ArrayBlockingQueue<String> q = new ArrayBlockingQueue<String>( 1000 );
        List<JStatParser> parsers = new ArrayList<JStatParser>( );

        BufferedReader reader = null;

        try
        {
            if( args.length == 2 )
            {
                reader = new BufferedReader( new FileReader( args[1] ) );
            }
            else
            {
                reader = new BufferedReader( new InputStreamReader( System.in ) );
            }

            for( int i = 0; i < numParsers; i++ )
            {
                JStatParser parser = new JStatParser( q );
                parser.start( );
                parsers.add( parser );
            }

            String line = null;
            while( (line = reader.readLine( )) != null )
            {
                try
                {
                    q.put( line );
                }
                catch( InterruptedException e )
                {
                    // TODO Auto-generated catch block
                    System.err.println( "Failed to add line to q: "
                            + e.getMessage( ) );
                    e.printStackTrace( );
                }
            }

            // At this point, we've put everything on the queue, now we just
            // need to wait for it to be processed.
            while( q.size( ) > 0 )
            {
                try
                {
                    Thread.sleep( 250 );
                }
                catch( InterruptedException e )
                {
                }
            }

            Map<String,Integer> countMap = new Hashtable<String,Integer>( );
            for( JStatParser jsp : parsers )
            {
                jsp.alldone( );
                Map<String,Integer> cm = jsp.getCountMap( );
                for( String key : cm.keySet( ) )
                {
                    if( countMap.containsKey( key ))
                    {
                        Integer currentCount = countMap.get(  key );
                        currentCount += cm.get( key );
                        countMap.put( key, currentCount );
                    }
                    else
                    {
                        countMap.put(  key, cm.get( key ) );
                    }
                }
            }

            System.out.println( "ID,Count" );
            for( String key : new TreeSet<String>(countMap.keySet( ))  )
            {
                System.out.println( key + "," + countMap.get( key ) );
            }

            for( JStatParser parser : parsers )
            {
                try
                {
                    parser.join( 100 );
                }
                catch( InterruptedException e )
                {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }

            System.exit(  0  );
        }
        catch( IOException e )
        {
            System.err.println( "Caught exception: " + e.getMessage( ) );
            e.printStackTrace( );
        }
    }
}

Answer 1

我不确定这对演员来说是个不错的测试案例.首先,演员之间几乎没有互动.这是一个简单的map/reduce,它需要并行性,而不是并发性.

演员的开销也很大,我不知道分配了多少实际线程.根据您拥有的处理器数量,您可能拥有比Java程序更少的线程 - 考虑到加速是4倍而不是5倍,这似乎是这种情况.

你编写演员的方式是针对空闲的演员进行优化的,那种情况你有数百或数千或演员,但只有少数人随时都在做实际的工作.如果用while/ receive代替loop/ 编写演员react,他们的表现会更好.

现在,演员可以轻松地在许多计算机上分发应用程序,除非您违反了演员的原则之一:您正在调用actor对象上的方法.你永远不应该与演员这样做,事实上,Akka阻止你这样做.执行此操作的更多方式是聚合器向每个actor询问其键集,计算它们的并集,然后,对于每个键,要求所有actor发送其对该键的计数.

但是,我不确定演员的开销是你所看到的.您没有提供有关Java实现的信息,但我敢说您使用可变映射,甚至可能使用单个并发可变映射 - 这与您在Scala中执行的实现完全不同.

还没有关于如何读取文件的信息(这样一个大文件可能有缓冲问题),或者如何在Java中解析它.由于大多数工作是读取和解析文件,而不是计算令牌,因此在实现方面的差异可以轻松克服任何其他问题.

最后,关于驻留内存大小,Scala有一个9 MB的库(除了JVM带来的),这可能就是你所看到的.当然,如果你在Java中使用单个并发映射与在Scala中使用6个不可变映射,那肯定会对内存使用模式产生很大影响.