Skip to content

feat: support fury serializer #356

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -87,6 +87,7 @@
<slf4j.version>1.7.21</slf4j.version>
<sofa.common.tools>1.4.0</sofa.common.tools>
<sortpom.maven.plugin>2.4.0</sortpom.maven.plugin>
<fury.version>0.6.0</fury.version>
</properties>

<dependencies>
Expand All @@ -96,6 +97,12 @@
<artifactId>slf4j-api</artifactId>
<version>${slf4j.version}</version>
</dependency>
<dependency>
<groupId>org.apache.fury</groupId>
<artifactId>fury-core</artifactId>
<version>${fury.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -34,12 +34,8 @@
public class HessianSerializer implements Serializer {

private SerializerFactory serializerFactory = new SerializerFactory();
private static ThreadLocal<ByteArrayOutputStream> localOutputByteArray = new ThreadLocal<ByteArrayOutputStream>() {
@Override
protected ByteArrayOutputStream initialValue() {
return new ByteArrayOutputStream();
}
};
private static ThreadLocal<ByteArrayOutputStream> localOutputByteArray = ThreadLocal.withInitial(
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

bolt用于多个基础组件,最好不要用高版本的用法保持线下兼容

ByteArrayOutputStream::new);

/**
* @see com.alipay.remoting.serialization.Serializer#serialize(java.lang.Object)
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package com.alipay.remoting.serialization;

import java.util.concurrent.locks.ReentrantLock;
import com.alipay.remoting.serialization.fury.FurySerializer;

/**
* Manage all serializers.
Expand All @@ -32,19 +33,27 @@ public class SerializerManager {

private static Serializer[] serializers = new Serializer[5];
public static final byte Hessian2 = 1;

//public static final byte Json = 2;

public static final byte Fury = 3;

private static final ReentrantLock REENTRANT_LOCK = new ReentrantLock();

public static Serializer getSerializer(int idx) {
Serializer currentSerializer = serializers[idx];
if (currentSerializer == null && idx == Hessian2) {
if (currentSerializer == null) {
REENTRANT_LOCK.lock();
try {
currentSerializer = serializers[idx];
if (currentSerializer == null) {
currentSerializer = new HessianSerializer();
addSerializer(Hessian2, currentSerializer);
if (idx == Hessian2) {
currentSerializer = new HessianSerializer();
addSerializer(Hessian2, currentSerializer);
} else if (idx == Fury) {
currentSerializer = new FurySerializer();
addSerializer(Fury, currentSerializer);
}
}
} finally {
REENTRANT_LOCK.unlock();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alipay.remoting.serialization.fury;

import java.util.ArrayList;
import java.util.List;
import com.alipay.remoting.exception.CodecException;
import com.alipay.remoting.serialization.Serializer;
import org.apache.fury.Fury;
import org.apache.fury.ThreadLocalFury;
import org.apache.fury.ThreadSafeFury;

/**
* @author [email protected]
*/
public class FurySerializer implements Serializer {

private static final List<Class<?>> REGISTRY_LIST = new ArrayList<>();

private final ThreadSafeFury fury = new ThreadLocalFury(classLoader -> {
Fury fury = Fury.builder().withRefTracking(true)
.requireClassRegistration(true).withClassLoader(classLoader).build();
REGISTRY_LIST.forEach(fury::register);
return fury;
});

@Override
public byte[] serialize(Object obj) throws CodecException {
try {
return fury.serialize(obj);
} catch (Exception e) {
throw new CodecException("Fury serialization failed", e);
}
}

@Override
public <T> T deserialize(byte[] data, String classOfT) throws CodecException {
return (T)fury.deserialize(data);
}
Comment on lines +50 to +53
Copy link

@coderabbitai coderabbitai bot Jul 31, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add type safety to deserialize method.

The deserialize method uses unchecked casting which can lead to ClassCastException. Consider adding type safety.

-    public <T> T deserialize(byte[] data, String classOfT) throws CodecException {
-        return (T)fury.deserialize(data);
+    public <T> T deserialize(byte[] data, Class<T> classOfT) throws CodecException {
+        return classOfT.cast(fury.deserialize(data));
Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
@Override
public <T> T deserialize(byte[] data, String classOfT) throws CodecException {
return (T)fury.deserialize(data);
}
@Override
public <T> T deserialize(byte[] data, Class<T> classOfT) throws CodecException {
return classOfT.cast(fury.deserialize(data));

Copy link
Contributor Author

@funky-eyes funky-eyes Jul 31, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

这个问题我无法决定是否如此修改,因为会涉及改动上层序列化接口

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@funky-eyes, I understand that changing the deserialize method's signature might involve modifications to the upper-level serialization interface and other dependent classes. The method is used in several places across the codebase, including FurySerializerTest.java, HessianSerializerTest.java, RpcResponseCommand.java, and more.

Given the widespread usage, this change would require careful consideration and testing. It might be best to discuss this further or address it in a separate PR to ensure all dependencies are handled properly. Let me know how you would like to proceed.

Analysis chain
Scripts executed

The following scripts were executed for the analysis:

Script:

#!/bin/bash
# Search for all invocations of the `deserialize` method in the codebase to understand its usage and potential impact of changing its signature.
rg --context 5 'deserialize\('

Length of output: 18313


public static void registry(Class<?> clazz) {
REGISTRY_LIST.add(clazz);
}

}
Original file line number Diff line number Diff line change
@@ -0,0 +1,101 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package com.alipay.remoting.serialization.fury;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.alipay.remoting.serialization.HessianSerializerTest;

import com.alipay.remoting.exception.CodecException;
import org.apache.fury.exception.InsecureException;
import org.junit.Assert;
import org.junit.Test;


import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;

/**
* @author [email protected]
*/
public class FurySerializerTest{

public static FurySerializer serializer;

static {
FurySerializer.registry(HessianSerializerTest.class);
serializer = new FurySerializer();
}

@Test
public void concurrentSerializeTest() throws InterruptedException {
int concurrentNum = 10;
CountDownLatch countDownLatch = new CountDownLatch(concurrentNum);
for (int i = 0; i < concurrentNum; ++i) {
FurySerializerTest.MyThread thread = new FurySerializerTest.MyThread(countDownLatch);
new Thread(thread).start();
}
countDownLatch.await(2, TimeUnit.SECONDS);

}

@Test
public void testSerializeError() {
FurySerializerTest furySerializerTest = new FurySerializerTest();
try {
serializer.serialize(furySerializerTest);
} catch (CodecException e) {
Assert.assertEquals(e.getCause().getClass(), InsecureException.class);
}
}

@Test
public void testSerialize() {
HessianSerializerTest furySerializerTest = new HessianSerializerTest();
try {
Assert.assertNotNull(serializer.serialize(furySerializerTest));
} catch (CodecException e) {
fail();
}
}

static class MyThread implements Runnable {
CountDownLatch countDownLatch;

public MyThread(CountDownLatch countDownLatch) {
this.countDownLatch = countDownLatch;
}

@Override
public void run() {
try {
for (int i = 0; i < 100; i++) {
String randomStr = UUID.randomUUID().toString();
byte[] bytes = serializer.serialize(randomStr);
String o = serializer.deserialize(bytes, String.class.getName());
assertEquals(o, randomStr);
}
} catch (Exception e) {
fail();
} finally {
countDownLatch.countDown();
}
}
}

}
Loading