用Akka构建一个简易的分布式文件系统

原本初期打算用Hadoop 2,但是后来有限的服务器部署了Solr Cloud,各类站点,发现资源不够了,近10T的文件,已经几乎把服务器的磁盘所有用光。想来想去,因为目前架构基于Scala的,因此仍是用Scala Akka实现了一个简单版本的分布式文件系统。html

Scala版本是2.10.3:http://www.scala-lang.org,Akka版本是2.2.3:http://akka.io。数据库

全部文件随机放在不一样的服务器上,在数据库中记录了文件存放的服务器IP地址、文件路径。在服务端部署基于Akka的简单文件服务,接收文件路径,读取并返回文件内容。调用者根据文件地址,去数据库中查找文件的服务IP地址和文件路径,根据获得的服务器IP地址,传入文件路径,调用该服务器的文件服务。服务器

如下是部分实现代码。架构

1.文件服务参数app

1 case class PatentFulltextArgs(
2   val url: String,
3   val start: Int,
4   val size: Int) {
5 
6 }

2.文件服务Trait(有点像WCF中的服务契约)tcp

1 trait PatentFulltextService {
2   def find(args: PatentFulltextArgs): Array[Byte]
3 }

3.文件服务实现分布式

 1 class PatentFulltextServiceImpl extends PatentFulltextService with Disposable {
 2   def find(args: PatentFulltextArgs): Array[Byte] = {
 3     val list = ListBuffer[Byte]()
 4     val file = FileSystems.getDefault().getPath(args.url)
 5 
 6     using(Files.newInputStream(file)) { in =>
 7       {
 8         val bytes = new Array[Byte](args.size + 1)
 9         in.skip(args.start)
10         in.read(bytes, 0, bytes.length)
11 
12         list ++= bytes
13       }
14     }
15 
16     list.toArray
17   }
18 }

4.用户Akka Deploy发布的类ide

class ServiceApplication extends Bootable {
  val system = ActorSystem("serivce", ConfigFactory.load.getConfig("service"))
  def startup() {
    TypedActor(system).typedActorOf(TypedProps[PatentFulltextServiceImpl], "patentfulltext")
  }

  def shutdown() {
    system.shutdown
  }
}

在这里,我使用的Akka的TypeActor,请参考:http://doc.akka.io/docs/akka/2.2.3/scala/typed-actors.html。oop

如下是部署过程。this

把生成的jar包,发布在Akka的deploy目录下,根据须要修改Akka的配置文件目录config下的application.conf。如下是我配置的内容,仅供参考:

actor {

provider = "akka.remote.RemoteActorRefProvider"

 

typed {

# Default timeout for typed actor methods with non-void return type

timeout = 6000s

}

}

remote {

transport = "akka.remote.netty.NettyRemoteTransport"

netty.tcp {

  hostname = "服务端IP"

  port = 2552

}

客户端使用时只须要服务契约Trait和相关实体类,如下是我写的一个客户端调用的类,仅供参考:

 1 object RemoteService {
 2   val logger = LoggerFactory.getLogger(this.getClass())
 3   private var system: ActorSystem = null
 4 
 5   def apply(configFile: String) = {
 6     system = ActorSystem("RemoteService", ConfigFactory.parseFile(new File(configFile)))
 7   }
 8 
 9   def findPatentFulltext(serverIp: String, patentFulltextArgs: PatentFulltextArgs) = {
10     TypedActor(system).typedActorOf(TypedProps[com.cloud.akka.service.model.PatentFulltextService], system.actorFor("akka.tcp://serivce@" + serverIp + ":2552/user/patentfulltext")).find(patentFulltextArgs)
11 
12   }
13 
14   def shutdown = {
15     if (null != system) system.shutdown()
16   }
17 }}

如下问题是我还没找到合适的解决办法:

1.Akka没法传输大文件,即便修改配置,服务器能够返回,可是接收的客户端还会报错。个人解决方案是在客户端分块读取,而后合并。

2.在客户端使用时,TypedActor没有找到使用ActorSelection构建,由于ActorFor是标记为Deprecated。

相关文章
相关标签/搜索